概述

上一篇文章中我们介绍了在RocksDB中如何在内存中查找对应的数据，这一篇我们将会详细介绍当内存中的数据不存在时，RocksDB如何在磁盘上查找对应的数据.

源码分析

依旧是从DBImpl::GetImpl开始，上一篇文章中我们分析这个函数只分析了Memtable相关的代码，这次我们来看当memtable没有查找到之后，RocksDB是如何处理的.我们可以看到当MemTable中没有找到对应的数据之后(包括删除)，RocksDB将会进入对于sst的查找.

  1.   if (!done) {
  2.     PERF_TIMER_GUARD(get_from_output_files_time);
  3.     sv->current->Get(read_options, lkey, pinnable_val, &s, &merge_context,
  4.                      &range_del_agg, value_found, nullptr, nullptr, callback,
  5.                      is_blob_index);
  6.     RecordTick(stats_, MEMTABLE_MISS);
  7.   }

从上面的代码我们可以看到直接从当前的version(sv->current)调用Get方法，因此接下来我们就来详细看这个函数。 这个函数简单来说就是根据所需要查找的key,然后选择对应的文件,这里每次会返回一个文件(key在sst的key范围内),然后循环查找.

先来看查找之前的初始化

  1.   GetContext get_context(
  2.       user_comparator(), merge_operator_, info_log_, db_statistics_,
  3.       status->ok() ? GetContext::kNotFound : GetContext::kMerge, user_key,
  4.       value, value_found, merge_context, range_del_agg, this->env_, seq,
  5.       merge_operator_ ? &pinned_iters_mgr : nullptr, callback, is_blob);
  7.   // Pin blocks that we read to hold merge operands
  8.   if (merge_operator_) {
  9.     pinned_iters_mgr.StartPinning();
  10.   }
  12.   FilePicker fp(
  13.       storage_info_.files_, user_key, ikey, &storage_info_.level_files_brief_,
  14.       storage_info_.num_non_empty_levels_, &storage_info_.file_indexer_,
  15.       user_comparator(), internal_comparator());
  16.   FdWithKeyRange* f = fp.GetNextFile();

第一个是GetContext结构，这个类只要是根据传递进来的文件元信息来查找对应的key.然后是FilePicker,这个类主要是根据传递进来的key来选择对应的文件.这里最重要就是GetNextFile这个函数，我们来看这个函数。

这个函数他会遍历所有的level,然后再遍历每个level的所有的文件,这里会对level 0的文件做一个特殊处理，这是因为只有level0的sst的range不是有序的，因此我们每次查找需要查找所有的文件，也就是会一个个的遍历.

而在非level0,我们只需要按照二分查找来得到对应的文件即可,如果二分查找不存在，那么我就需要进入下一个level进行查找.

  1. FdWithKeyRange* GetNextFile() {
  2.     while (!search_ended_) {  // Loops over different levels.
  3.       while (curr_index_in_curr_level_ < curr_file_level_->num_files) {
  4.         // Loops over all files in current level.
  5.         FdWithKeyRange* f = &curr_file_level_->files[curr_index_in_curr_level_];
  6.         hit_file_level_ = curr_level_;
  7.         is_hit_file_last_in_level_ =
  8.             curr_index_in_curr_level_ == curr_file_level_->num_files - 1;
  9.         int cmp_largest = -1;
  10.         if (num_levels_ > 1 || curr_file_level_->num_files > 3) {
  11.           // Check if key is within a file's range. If search left bound and
  12.           // right bound point to the same find, we are sure key falls in
  13.           // range.
  14.           assert(
  15.               curr_level_ == 0 ||
  16.               curr_index_in_curr_level_ == start_index_in_curr_level_ ||
  17.               user_comparator_->Compare(user_key_,
  18.                 ExtractUserKey(f->smallest_key)) <= 0);
  20.           int cmp_smallest = user_comparator_->Compare(user_key_,
  21.               ExtractUserKey(f->smallest_key));
  22.           if (cmp_smallest >= 0) {
  23.             cmp_largest = user_comparator_->Compare(user_key_,
  24.                 ExtractUserKey(f->largest_key));
  25.           }
  27.           // Setup file search bound for the next level based on the
  28.           // comparison results
  29.           if (curr_level_ > 0) {
  30.             file_indexer_->GetNextLevelIndex(curr_level_,
  31.                                             curr_index_in_curr_level_,
  32.                                             cmp_smallest, cmp_largest,
  33.                                             &search_left_bound_,
  34.                                             &search_right_bound_);
  35.           }
  36.           // Key falls out of current file's range
  37.           if (cmp_smallest < 0 || cmp_largest > 0) {
  38.             if (curr_level_ == 0) {
  39.               ++curr_index_in_curr_level_;
  40.               continue;
  41.             } else {
  42.               // Search next level.
  43.               break;
  44.             }
  45.           }
  46.         }
  47.         returned_file_level_ = curr_level_;
  48.         if (curr_level_ > 0 && cmp_largest < 0) {
  49.           // No more files to search in this level.
  50.           search_ended_ = !PrepareNextLevel();
  51.         } else {
  52.           ++curr_index_in_curr_level_;
  53.         }
  54.         return f;
  55.       }
  56.       // Start searching next level.
  57.       search_ended_ = !PrepareNextLevel();
  58.     }
  59.     // Search ended.
  60.     return nullptr;
  61.   }

这里RocksDB使用了一个技巧用来加快二分查找的速度，每次更新sst的时候，RocksDB都会调用FileIndexer::UpdateIndex来更新这样的一个结构,这个结构就是FileIndexer，它主要是用来保存每一个level和level+1的key范围的关联信息，这样当我们在level查找的时候，如果没有查找到信息，那么我们将会迅速得到下一个level需要查找的文件范围.每一个key来进行比较总会有三种情况:

  • 小于当前sst的smallest.
  • 大于当前sst的largest.
  • 处于这个范围.

那么我们只需要在初始化索引的时候能够得到当前的sst在下一个level中的位置，就可以根据上面三种类型来确定下一个level我们需要进行二分查找的文件范围.在RocksDB中定义了下面三个值.

  1.     // Point to a left most file in a lower level that may contain a key,
  2.     // which compares greater than smallest of a FileMetaData (upper level)
  3.     int32_t smallest_lb;
  4.     // Point to a left most file in a lower level that may contain a key,
  5.     // which compares greater than largest of a FileMetaData (upper level)
  6.     int32_t largest_lb;
  7.     // Point to a right most file in a lower level that may contain a key,
  8.     // which compares smaller than smallest of a FileMetaData (upper level)
  9.     int32_t smallest_rb;
  10.     // Point to a right most file in a lower level that may contain a key,
  11.     // which compares smaller than largest of a FileMetaData (upper level)
  12.     int32_t largest_rb;

我们通过例子来解释这三个值.假设有下面两个level,4个sst.那么初始化的时候，对应的level1的这个sst对应的四个值分别为. smallest_lb=１;largest_lb=2;smallest_rb=1;largest_rb=2;

  1.         level 1:              [50 - 60]
  2.         level 2:        [1 - 40], [45 - 55], [58 - 80]

此时如果我们查找一个key为49,然后第一次比较，也就是key < level1.sst->smallest,那么我们将会知道我们需要在0和smallest_rb之间来查找,也就是0和1.假设我们查找key是55,也就是 level1.sst->smallest < key < level1.test.largest,此时我们在level2将需要在smallest_rb和largest_rb之间.这里可以看到其实就是计算一个重合的区间。

来看RocksDB如何根据当前level的比较结果来计算下一个level需要二分查找的文件范围:

  1. void FileIndexer::GetNextLevelIndex(const size_t level, const size_t file_index,
  2.                                     const int cmp_smallest,
  3.                                     const int cmp_largest, int32_t* left_bound,
  4.                                     int32_t* right_bound) const {
  5.   assert(level > 0);
  7.   const IndexUnit* index_units = next_level_index_[level].index_units;
  8.   const auto& index = index_units[file_index];
  10.   if (cmp_smallest < 0) {
  11.     *left_bound = (level > 0 && file_index > 0)
  12.                       ? index_units[file_index - 1].largest_lb
  13.                       : 0;
  14.     *right_bound = index.smallest_rb;
  15.   } else if (cmp_smallest == 0) {
  16.     *left_bound = index.smallest_lb;
  17.     *right_bound = index.smallest_rb;
  18.   } else if (cmp_smallest > 0 && cmp_largest < 0) {
  19.     *left_bound = index.smallest_lb;
  20.     *right_bound = index.largest_rb;
  21.   } else if (cmp_largest == 0) {
  22.     *left_bound = index.largest_lb;
  23.     *right_bound = index.largest_rb;
  24.   } else if (cmp_largest > 0) {
  25.     *left_bound = index.largest_lb;
  26.     *right_bound = level_rb_[level + 1];
  27.   } else {
  28.     assert(false);
  29.   }
  30. }

看完上面这些我们继续来看RocksDB对于文件的查找.这里所有对于key的查找都是在table_cache_->Get中.这里我们暂且略过这个函数的实现，最后我们再来详细分析这个函数.

  1.  while (f != nullptr) {
  2. ................................
  4.     *status = table_cache_->Get(
  5.         read_options, *internal_comparator(), f->fd, ikey, &get_context,
  6.         cfd_->internal_stats()->GetFileReadHist(fp.GetHitFileLevel()),
  7.         IsFilterSkipped(static_cast<int>(fp.GetHitFileLevel()),
  8.                         fp.IsHitFileLastInLevel()),
  9.         fp.GetCurrentLevel());
  10.     // TODO: examine the behavior for corrupted key
  11.     if (!status->ok()) {
  12.       return;
  13.     }
  14. .......................
  15.  }

当table_cache_->Get返回之后，我们需要根据get_context来判断返回的结果

  2. switch (get_context.State()) {
  3.       case GetContext::kNotFound:
  4.         // Keep searching in other files
  5.         break;
  6.       case GetContext::kMerge:
  7.         break;
  8.       case GetContext::kFound:
  9.         if (fp.GetHitFileLevel() == 0) {
  10.           RecordTick(db_statistics_, GET_HIT_L0);
  11.         } else if (fp.GetHitFileLevel() == 1) {
  12.           RecordTick(db_statistics_, GET_HIT_L1);
  13.         } else if (fp.GetHitFileLevel() >= 2) {
  14.           RecordTick(db_statistics_, GET_HIT_L2_AND_UP);
  15.         }
  16.         return;
  17.       case GetContext::kDeleted:
  18.         // Use empty error message for speed
  19.         *status = Status::NotFound();
  20.         return;
  21.       case GetContext::kCorrupt:
  22.         *status = Status::Corruption("corrupted key for ", user_key);
  23.         return;
  24.       case GetContext::kBlobIndex:
  25.         ROCKS_LOG_ERROR(info_log_, "Encounter unexpected blob index.");
  26.         *status = Status::NotSupported(
  27.             "Encounter unexpected blob index. Please open DB with "
  28.             "rocksdb::blob_db::BlobDB instead.");
  29.         return;
  30.     }

如果没有发现对应的值则进入下一次文件查找

  1. f = fp.GetNextFile();

最后我们来详细分析最核心的函数TableCache::Get,这个函数不仅仅是返回对应的查找结果，并且还会cache相应的文件信息，并且如果row_cache打开，他还会做row cache.这里row cache就是对当前的所需要查找的key在当前sst中对应的value进行cache.

先来看如果打开了row cache,RocksDB将会如何处理,首先它会计算row cache的key.通过下面的代码我们可以看到row cache的key就是fd_number+seq_no+user_key.

  1.     uint64_t fd_number = fd.GetNumber();
  2.     auto user_key = ExtractUserKey(k);
  3.     // We use the user key as cache key instead of the internal key,
  4.     // otherwise the whole cache would be invalidated every time the
  5.     // sequence key increases. However, to support caching snapshot
  6.     // reads, we append the sequence number (incremented by 1 to
  7.     // distinguish from 0) only in this case.
  8.     uint64_t seq_no =
  9.         options.snapshot == nullptr ? 0 : 1 + GetInternalKeySeqno(k);
  11.     // Compute row cache key.
  12.     row_cache_key.TrimAppend(row_cache_key.Size(), row_cache_id_.data(),
  13.                              row_cache_id_.size());
  14.     AppendVarint64(&row_cache_key, fd_number);
  15.     AppendVarint64(&row_cache_key, seq_no);
  16.     row_cache_key.TrimAppend(row_cache_key.Size(), user_key.data(),
  17.                              user_key.size());

然后就是在row cache中进行一次查找.如果有对应的值则直接返回结果，否则则将会在对应的sst读取传递进来的key.

  1.     if (auto row_handle =
  2.             ioptions_.row_cache->Lookup(row_cache_key.GetUserKey())) {
  3.       Cleanable value_pinner;
  4.       auto release_cache_entry_func = [](void* cache_to_clean,
  5.                                          void* cache_handle) {
  6.         ((Cache*)cache_to_clean)->Release((Cache::Handle*)cache_handle);
  7.       };
  8.       auto found_row_cache_entry = static_cast<const std::string*>(
  9.           ioptions_.row_cache->Value(row_handle));
  10. ....................................
  11.       done = true;       
  12.     } else {
  13.       // Not found, setting up the replay log.
  14.       RecordTick(ioptions_.statistics, ROW_CACHE_MISS);
  15.       row_cache_entry = &row_cache_entry_buffer;
  16.     }

接下来就是需要在对应的sst文件读取对应的key的值,这里可以看到每一个fd都包含了一个TableReader的结构，这个结构就是用来保存文件的内容.而我们的table_cache主要就是缓存这个结构.

  1.  Status s;
  2.   TableReader* t = fd.table_reader;
  3.   Cache::Handle* handle = nullptr;
  4.   if (!done && s.ok()) {
  5.     if (t == nullptr) {
  6.       s = FindTable(env_options_, internal_comparator, fd, &handle,
  7.                     options.read_tier == kBlockCacheTier /* no_io */,
  8.                     true /* record_read_stats */, file_read_hist, skip_filters,
  9.                     level);
  10.       if (s.ok()) {
  11.         t = GetTableReaderFromHandle(handle);
  12.       }
  13.     }
  14.    ..........................
  15.   }

上面的代码会直接调用TableCache::FindTable, 这个函数主要是用来实现对应tablereader的读取以及row cache.

  1. Status TableCache::FindTable(const EnvOptions& env_options,
  2.                              const InternalKeyComparator& internal_comparator,
  3.                              const FileDescriptor& fd, Cache::Handle** handle,
  4.                              const bool no_io, bool record_read_stats,
  5.                              HistogramImpl* file_read_hist, bool skip_filters,
  6.                              int level,
  7.                              bool prefetch_index_and_filter_in_cache) {
  8. ...................................................
  9.   if (*handle == nullptr) {
  10.     if (no_io) {  // Don't do IO and return a not-found status
  11.       return Status::Incomplete("Table not found in table_cache, no_io is set");
  12.     }
  13.     unique_ptr<TableReader> table_reader;
  14.     s = GetTableReader(env_options, internal_comparator, fd,
  15.                        false /* sequential mode */, 0 /* readahead */,
  16.                        record_read_stats, file_read_hist, &table_reader,
  17.                        skip_filters, level, prefetch_index_and_filter_in_cache);
  18.     if (!s.ok()) {
  19.       assert(table_reader == nullptr);
  20.       RecordTick(ioptions_.statistics, NO_FILE_ERRORS);
  21.       // We do not cache error results so that if the error is transient,
  22.       // or somebody repairs the file, we recover automatically.
  23.     } else {
  24.       s = cache_->Insert(key, table_reader.get(), 1, &DeleteEntry<TableReader>,
  25.                          handle);
  26.       if (s.ok()) {
  27.         // Release ownership of table reader.
  28.         table_reader.release();
  29.       }
  30.     }
  31.   }
  32.   return s;
  33. }

通过上面的代码可以看到实现很简单，就是一般的cache逻辑，读取然后判断是否存在，不存在则插入到cache. 上面的函数会调用 TableCache::GetTableReader，我们来简单看下这个函数.

  1. Status TableCache::GetTableReader(
  2.     const EnvOptions& env_options,
  3.     const InternalKeyComparator& internal_comparator, const FileDescriptor& fd,
  4.     bool sequential_mode, size_t readahead, bool record_read_stats,
  5.     HistogramImpl* file_read_hist, unique_ptr<TableReader>* table_reader,
  6.     bool skip_filters, int level, bool prefetch_index_and_filter_in_cache,
  7.     bool for_compaction) {
  8. ..........................................
  9.   if (s.ok()) {
  10. ...............................................    
  11.     s = ioptions_.table_factory->NewTableReader(
  12.         TableReaderOptions(ioptions_, env_options, internal_comparator,
  13.                            skip_filters, level),
  14.         std::move(file_reader), fd.GetFileSize(), table_reader,
  15.         prefetch_index_and_filter_in_cache);
  16.     TEST_SYNC_POINT("TableCache::GetTableReader:0");
  17.   }
  18.   return s;
  19. }

可以看到最关键的调用就是调用ioptions_.table_factory->NewTableReader,　这里RocksDB会根据我们配置的不同的sst格式来调用不同的reader,而在RocksDB中默认的格式是基于block.

  1. // Create default block based table factory.
  2. extern TableFactory* NewBlockBasedTableFactory(
  3.     const BlockBasedTableOptions& table_options = BlockBasedTableOptions());

这里我们就不详细分析sst的文件格式了，以后我们会来详细对比这几个文件格式的优劣.这里我们只需要知道最终缓存的tablereader就是一个BlockBasedTable对象(假设使用了基于block的sst format).

当读取完毕TableReader之后，RocksDB就需要从sst文件中get key了,也就是最终的key查找方式是在每个sst format class的Get方法中实现的。

  1.     if (s.ok()) {
  2.       get_context->SetReplayLog(row_cache_entry);  // nullptr if no cache.
  3.       s = t->Get(options, k, get_context, skip_filters);
  4.       get_context->SetReplayLog(nullptr);
  5.     }

和上面一样，这里的get也就是对应的sst format的get.

最后如果查找到key,则开始缓存对应的kv到row_cache.

  1.     size_t charge =
  2.         row_cache_key.Size() + row_cache_entry->size() + sizeof(std::string);
  3.     void* row_ptr = new std::string(std::move(*row_cache_entry));
  4.     ioptions_.row_cache->Insert(row_cache_key.GetUserKey(), row_ptr, charge,
  5.                                 &DeleteEntry<std::string>);

这里整个读取流程我们都分析完毕了，不过这里略过了merge,delete range以及不同sst format如何组织以及读取内容，后续我们会详细分析这些略过的内容.