leveldb代码阅读（8）——查询数据

1、DBImpl::Put 函数用于数据查询

2、流程如下：

3、首先在可读可写的内存table中查询，查询到就返回

4、在只读内存table中查询，查询到就返回

5、如果都没有找到，那么只能在硬盘中查询了，这是一个比较复杂的过程，通过Version::Get函数实现。

Version::Get函数的流程：

1、主要的功能就是从上到下在每一个level中进行查询数据，如果找到就返回

2、遍历每一个level

3、对于level 0,比较特别，因为level0的数据直接由内存表转储得到，各个SSTable之间可能会有重复的数据。所以要遍历level 0的每一个文件，然后找到匹配的数据，存放到一个列表中，然后对列表进行排序（最新的数据排在最前面），那么列表的第一个数据就是要找查询的数据

4、对于其他的level，因为经过了compact阶段，所以要查询的数据只可能存在与一个文件中，又因为大于0的level的键值都是排序好的，所以很容易就可以找到数据所在的文件，找到文件之后，就在文件内部进行查找。

5、数据查找的阶段是在文件的元数据中进行的，找到对应的文件之后，还要在table 的cache中获取这个数据，如果没有，要把它从硬盘中读取到内存中，然后返回查找结果

// 读取数据（查询数据）
Status DBImpl::Get(const ReadOptions& options,
const Slice& key,
std::string* value)
{
Status s;
MutexLock l(&mutex_);
// 记录快照的编号
SequenceNumber snapshot;
if (options.snapshot != NULL) {
snapshot = reinterpret_cast<const SnapshotImpl*>(options.snapshot)->number_;
} else {
snapshot = versions_->LastSequence();
}

MemTable* mem = mem_;
MemTable* imm = imm_;
Version* current = versions_->current();
mem->Ref();
if (imm != NULL) imm->Ref();
current->Ref();

bool have_stat_update = false;
Version::GetStats stats;

// Unlock while reading from files and memtables
{
mutex_.Unlock();
// First look in the memtable, then in the immutable memtable (if any).

// 查找键
LookupKey lkey(key, snapshot);
// 在可读可写的内存表中查询
if (mem->Get(lkey, value, &s)) {
// Done
}
// 在只读表中查询
else if (imm != NULL && imm->Get(lkey, value, &s)) {
// Done
}
// 在硬盘中查询
else {
s = current->Get(options, lkey, value, &stats);
// 设置状态需要更新选项
have_stat_update = true;
}
mutex_.Lock();
}

if (have_stat_update && current->UpdateStats(stats)) {
MaybeScheduleCompaction();
}
mem->Unref();
if (imm != NULL) imm->Unref();
current->Unref();
return s;
}

// 在版本当（的SSTable）中进行查询
Status Version::Get(const ReadOptions& options,
const LookupKey& k,
std::string* value,
GetStats* stats)
{
Slice ikey = k.internal_key();
Slice user_key = k.user_key();
const Comparator* ucmp = vset_->icmp_.user_comparator();
Status s;

stats->seek_file = NULL;
stats->seek_file_level = -1;
FileMetaData* last_file_read = NULL;
int last_file_read_level = -1;

// We can search level-by-level since entries never hop across
// levels.  Therefore we are guaranteed that if we find data
// in an smaller level, later levels are irrelevant.
std::vector<FileMetaData*> tmp;
FileMetaData* tmp2;

// 在每一层中进行查找
for (int level = 0; level < config::kNumLevels; level++) {
size_t num_files = files_[level].size();
if (num_files == 0) continue;

// Get the list of files to search in this level
FileMetaData* const* files = &files_[level][0];

// 对于第0层，比较特别
if (level == 0) {
// Level-0 files may overlap each other.  Find all files that
// overlap user_key and process them in order from newest to oldest.
tmp.reserve(num_files);

// 可能多个文件都有这个键，把它添加到列表中
for (uint32_t i = 0; i < num_files; i++) {
FileMetaData* f = files[i];
if (ucmp->Compare(user_key, f->smallest.user_key()) >= 0 &&
ucmp->Compare(user_key, f->largest.user_key()) <= 0) {
tmp.push_back(f);
}
}
if (tmp.empty()) continue;

std::sort(tmp.begin(), tmp.end(), NewestFirst);
files = &tmp[0];
num_files = tmp.size();
}
// 对于其他层
else {
// Binary search to find earliest index whose largest key >= ikey.
// 查找
uint32_t index = FindFile(vset_->icmp_, files_[level], ikey);
// 没有找到
if (index >= num_files) {
files = NULL;
num_files = 0;
}
// 找到了
else
{
tmp2 = files[index];
if (ucmp->Compare(user_key, tmp2->smallest.user_key()) < 0)
{
// All of "tmp2" is past any data for user_key
files = NULL;
num_files = 0;
} else {
files = &tmp2;
num_files = 1;
}
}
}

// 然后进行处理（files中保存了可能有待查找键的文件）
for (uint32_t i = 0; i < num_files; ++i)
{
if (last_file_read != NULL && stats->seek_file == NULL)
{
// We have had more than one seek for this read.  Charge the 1st file.
stats->seek_file = last_file_read;
stats->seek_file_level = last_file_read_level;
}

FileMetaData* f = files[i];
last_file_read = f;
last_file_read_level = level;

Saver saver;
saver.state = kNotFound;
saver.ucmp = ucmp;
saver.user_key = user_key;
saver.value = value;

// 在对应的表cache中查找
s = vset_->table_cache_->Get(options, f->number, f->file_size,
ikey, &saver, SaveValue);
if (!s.ok()) {
return s;
}
switch (saver.state) {
case kNotFound:
break;      // Keep searching in other files
case kFound:
return s;
case kDeleted:
s = Status::NotFound(Slice());  // Use empty error message for speed
return s;
case kCorrupt:
s = Status::Corruption("corrupted key for ", user_key);
return s;
}
}
}

return Status::NotFound(Slice());  // Use an empty error message for speed
}

// 在table的cache中查询数据，如果没有找到，需要把文件从硬盘中加载进来
Status TableCache::Get(const ReadOptions& options,
uint64_t file_number,
uint64_t file_size,
const Slice& k,
void* arg,
void (*saver)(void*, const Slice&, const Slice&))
{
Cache::Handle* handle = NULL;

// 在cache中查找对应的表
Status s = FindTable(file_number, file_size, &handle);
if (s.ok()) {
Table* t = reinterpret_cast<TableAndFile*>(cache_->Value(handle))->table;
// 查找数据
s = t->InternalGet(options, k, arg, saver);
cache_->Release(handle);
}
return s;
}

// 在cache中查找一个table
Status TableCache::FindTable(uint64_t file_number, // table文件的序号
uint64_t file_size, // 文件的大小
Cache::Handle** handle) // 处理句柄
{
Status s;
char buf[sizeof(file_number)];
EncodeFixed64(buf, file_number);
Slice key(buf, sizeof(buf));

// 查找这个table是否在cache中
// 如果不存在需要把它载入cache中
*handle = cache_->Lookup(key);
if (*handle == NULL)
{
// 载入table文件
std::string fname = TableFileName(dbname_, file_number);
RandomAccessFile* file = NULL;
Table* table = NULL;
s = env_->NewRandomAccessFile(fname, &file);
if (!s.ok())
{
std::string old_fname = SSTTableFileName(dbname_, file_number);
if (env_->NewRandomAccessFile(old_fname, &file).ok()) {
s = Status::OK();
}
}

// 载入成功，打开table
if (s.ok()) {
s = Table::Open(*options_, file, file_size, &table);
}

// 载入或者打开失败，那么删除文件
if (!s.ok())
{
assert(table == NULL);
delete file;
// We do not cache error results so that if the error is transient,
// or somebody repairs the file, we recover automatically.
}
// 成功，创建一个table和文件的映射实体
// 然后插入cache列表中
else
{
TableAndFile* tf = new TableAndFile;
tf->file = file;
tf->table = table;
*handle = cache_->Insert(key, tf, 1, &DeleteEntry);
}
}
return s;
}