typedef uint64_t SequenceNumber;
static const SequenceNumber kMaxSequenceNumber = ((0x1ull << 56) - 1);

enum ValueType { kTypeDeletion = 0x0, kTypeValue = 0x1 };

struct ParsedInternalKey {
    Slice user_key;
    SequenceNumber sequence;
    ValueType type;
}

int InternalKeyComparator::Compare(const Slice& akey, const Slice& bkey) const {
    // Order by:
    //    user key
    //    sequence number 倒序
    //    type 倒序，实际不需要，因为sequence number唯一的
    int r = user_comparator_->Compare(ExtractUserKey(akey), ExtractUserKey(bkey));
    if (r == 0) {
        const uint64_t anum = DecodeFixed64(akey.data() + akey.size() - 8);
        const uint64_t bnum = DecodeFixed64(bkey.data() + bkey.size() - 8);
        if (anum > bnum) {
            r = -1;
        } else if (anum < bnum) {
            r = +1;
        }
    }
    return r;
}

db/dbformat.h db/dbformat.cc: 定义了Internal Key、Lookup Key、InternalKeyComparator和InternalFilterPolicy

// db/db_impl.cc

Status DB::Put(const WriteOptions& opt, const Slice& key, const Slice& value) {
  WriteBatch batch;
  batch.Put(key, value);
  return Write(opt, &batch);
}

Status DB::Delete(const WriteOptions& opt, const Slice& key) {
  WriteBatch batch;
  batch.Delete(key);
  return Write(opt, &batch);
}

struct WriteBatch {
    uint64 sequence;
    uint32 count;
    Record record[count];
};

// db/db_impl.cc
// 用来封装一个WriteBatch，用来标识状态
struct DBImpl::Writer {
    explicit Writer(port::Mutex* mu): batch(nullptr), sync(false), done(false), cv(mu) {}
    Status status;        // 写入的状态
    WriteBatch* batch;    // 对应的WriteBatch
    bool sync;            // 写日志时，是否需要sync
    bool done;            // 写入是否要完成
    port::CondVar cv;     // 用来等待其它线程的通知
};

Status DBImpl::Write(const WriteOptions& options, WriteBatch* updates) {
    // step1：构造一个writer，插入到wrtiers_里，注意插入前需要先获取锁
    Writer w(&mutex_);
    w.batch = updates;
    w.sync = options.sync;
    w.done = false;
    MutexLock l(&mutex_);
    writers_.push_back(&w);

    // step2: 检查done判断是否完成，或者自己是否是writers_队列里的第一个成员。有可能其它线程写入时把自己的内容也写入了，
    // 这样自己就是done，或者当自己是头元素了，表示轮到自己写入了
    while (!w.done && &w != writers_.front()) {
        w.cv.Wait();
    }

    // step3: 判断写入是否完成了，完成了就可以返回了
    if (w.done) {
        return w.status;
    }

    // step4：如果写入太快，进行限流，如果MemTable满了，生成新的MemTable
    Status status = MakeRoomForWrite(updates == nullptr);
    if (status.ok() && updates != nullptr) {
        // step5：从writers_头部扫描足够多的WriteBatch构造一个WriteBatch updates，last_writer保存了最后一个写入的writer
        // 设定新的WriteBatch的SequenceNumber
        WriteBatch* updates = BuildBatchGroup(&last_writer);
        WriteBatchInternal::SetSequence(updates, last_sequence + 1);
        last_sequence += WriteBatchInternal::Count(updates);
        // 注意这一步解锁，很关键，因为接下来的写入可能是一个费时的过程，解锁后，其它线程可以Get，其它线程也可以继续将writer
        // 插入到writers_里面，但是插入后，因为不是头元素，会等待，所以不会冲突
        mutex_.Unlock();
        // step6: 写入log，根据选项sync
        status = log_->AddRecord(WriteBatchInternal::Contents(updates));
        if (status.ok() && options.sync) {
            status = logfile_->Sync();
        }
        // step7: 写入到MemTable里
        if (status.ok()) {
            status = WriteBatchInternal::InsertInto(updates, mem_);
        }
        // 加锁需要修改全局的SequenceNumber以及writers_
        mutex_.Lock();
        // 更新全局的SequenceNumber
        versions_->SetLastSequence(last_sequence);
    }

    // step8: 从writers_队列从头开始，将写入完成的writer标识成done，并且弹出，通知这些writer
    // 这样这些writer的线程会被唤醒，发现自己的写入已经完成了，就会返回
    while (true) {
        Writer* ready = writers_.front();
        writers_.pop_front();
        if (ready != &w) {
            ready->status = status;
            ready->done = true;
            ready->cv.Signal();
        }
        if (ready == last_writer) break;
    }
    // 如果writers_里还有元素，就通知头元素，让它可以进来开始写入
    if (!writers_.empty()) {
        writers_.front()->cv.Signal();
    }
    return status;
}

// db/db_impl.cc
WriteBatch* DBImpl::BuildBatchGroup(Writer** last_writer) {
    Writer* first = writers_.front();
    WriteBatch* result = first->batch;
    // 计算writebatch的最大size，如果第一个的size比较小的话，限制最大的size，以防小的写入太慢
    size_t size = WriteBatchInternal::ByteSize(first->batch);
    size_t max_size = 1 << 20;
    if (size <= (128 << 10)) {
        max_size = size + (128 << 10);
    }
    *last_writer = first;
    std::deque<Writer*>::iterator iter = writers_.begin();
    ++iter;
    for (; iter != writers_.end(); ++iter) {
        Writer* w = *iter;
        // 如果这是一个sync的写入，但是第一个元素不是sync的话，那么就结束了，因为整体的写入都不是sync的
        if (w->sync && !first->sync) {
            break;
        }
        if (w->batch != nullptr) {
            size += WriteBatchInternal::ByteSize(w->batch);
            // 如果太大，就结束
            if (size > max_size) {
                break;
            }
            // 追加到result
            if (result == first->batch) {
                result = tmp_batch_;
                WriteBatchInternal::Append(result, first->batch);
            }
            WriteBatchInternal::Append(result, w->batch);
        }
        // 更新最后一个writer
        *last_writer = w;
    }
    return result;
}

// db/db_impl.cc
Status DBImpl::Get(const ReadOptions& options, const Slice& key, std::string* value) {
    Status s;
    // 加锁，读取元数据，Ref操作
    MutexLock l(&mutex_);
    // 设定snapshot，其实就是一个SequenceNumber，这是实现Snapshot的关键，设置成选项中的snapshot，或者
    // 当前最近的SequenceNumber
    SequenceNumber snapshot;
    if (options.snapshot != nullptr) {
        snapshot = static_cast<const SnapshotImpl*>(options.snapshot)->sequence_number();
    } else {
        snapshot = versions_->LastSequence();
    }
    // mem是MemTable，imm是Immutable MemTable，
    // SSTable则由当前的version代表，一个version包含了当前版本的SSTable的集合
    MemTable* mem = mem_;
    MemTable* imm = imm_;
    Version* current = versions_->current();
    // 这里都调用了Ref，LevelDB里大量使用了这种引用计数的方式管理对象，这里表示读取需要引用这些对象，假设在读取过程
    // 中其他线程发生了MemTable写满了，或者Immutable MemTable写入完成了需要删除了，或者做了一次Compaction，生
    // 成了新的Version，所引用的SSTable不再有效了，这些都需要对这些对象做一些改变，比如删除等，但是当前线程还引用着
    // 这些对象，所以这些对象还不能被删除。采用引用计数，其它线程删除对象时只是简单的Unref，因为当前线程还引用着这些
    // 对象，所以计数>=1，这些对象不会被删除，而当读取结束，调用Unref时，如果对象的计数是0，那么对象会被删除。
    mem->Ref();
    if (imm != nullptr) imm->Ref();
    current->Ref();
    {
        // 实际读取时解锁
        mutex_.Unlock();
        // 构造一个Lookup Key搜索MemTable
        LookupKey lkey(key, snapshot);
        // 搜索MemTable
        if (mem->Get(lkey, value, &s)) {
        // 搜索Immutable
        } else if (imm != nullptr && imm->Get(lkey, value, &s)) {
        // 搜索SSTable
        } else {
            s = current->Get(options, lkey, value, &stats);
        }
        mutex_.Lock();
    }
    // Unref释放
    mem->Unref();
    if (imm != nullptr) imm->Unref();
    current->Unref();
    return s;
}

// db/memtable.cc
bool MemTable::Get(const LookupKey& key, std::string* value, Status* s) {
    // Lookup Key的内容，搜索的Key
    Slice memkey = key.memtable_key();
    // 构建迭代器搜索
    Table::Iterator iter(&table_);
    iter.Seek(memkey.data());
    if (iter.Valid()) {
        const char* entry = iter.key();
        uint32_t key_length;
        const char* key_ptr = GetVarint32Ptr(entry, entry + 5, &key_length);
        // 看查找到的键里面包含的User Key是否和搜索的User Key相同
        // 这里需要这个判断是因为迭代器Seek时，指向大于等于搜索键的位置，所以有可能这个键是大于搜索的键的
        // 这边不需要判断SequenceNumber，便可实现snapshot功能，原因是搜索的键里面是包含SequenceNumber
        // 的，并且User Key相同时，SequenceNumber大的排在前面。所以Seek时跳过了User Key相同，但是SequenceNumber
        // 大于当前搜索的键的SequenceNumber的键，所以找到的就是那个snapshot之前的状态。
        if (comparator_.comparator.user_comparator()->Compare(Slice(key_ptr, key_length - 8),  key.user_key()) == 0) {
            // User Key匹配
            const uint64_t tag = DecodeFixed64(key_ptr + key_length - 8);
            switch (static_cast<ValueType>(tag & 0xff)) {
                // 如果tag是一个插入，那么解析值，返回
                case kTypeValue: {
                    Slice v = GetLengthPrefixedSlice(key_ptr + key_length);
                    value->assign(v.data(), v.size());
                    return true;
                }
                // 如果tag是一个删除，表示键找不到返回
                case kTypeDeletion:
                    *s = Status::NotFound(Slice());
                    return true;
            }
        }
    }
    return false;
}

// db/version_edit.h
struct FileMetaData {
    FileMetaData() : refs(0), allowed_seeks(1 << 30), file_size(0) {}
    int refs;
    int allowed_seeks;     // 判断什么时候可以Compaction
    uint64_t number;       // 文件编号
    uint64_t file_size;    // 文件大小bytes
    InternalKey smallest;  // 这个SSTable里最小的Internal Key
    InternalKey largest;   // 这个SSTable里最大的Internal Key
};

std::vector<FileMetaData*> Version::files_[config::kNumLevels];

// db/version_set.cc
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;
    // 这边从Level 0开始，因为当在低Level发现了一个键后，可以忽略高Level的键
    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;
        // 开始搜索当前的level
        FileMetaData* const* files = &files_[level][0];
        // 对于Level 0，需要做特殊处理，因为每个文件都有可能包含键，所以需要逐个检查
        if (level == 0) {
            // 这里将符合键在smallest和largest之间的FileMetaData都加入tmp
            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 {
            // 对于其它的level，只需要通过二分搜索找到相应的文件。
            uint32_t index = FindFile(vset_->icmp_, files_[level], ikey);
            // 未找到
            if (index >= num_files) {
                files = nullptr;
                num_files = 0;
            } else {
                tmp2 = files[index];
                if (ucmp->Compare(user_key, tmp2->smallest.user_key()) < 0) {
                    // 这个文件里最小键大于user_key
                    files = nullptr;
                    num_files = 0;
                } else {
                    files = &tmp2;
                    num_files = 1;
                }
            }
        }
        // 经过以上步骤，files开始的num_files个文件就是需要搜索的SSTable，非Level 0层num_files=1
        // 开始逐个搜索SSTable
        for (uint32_t i = 0; i < num_files; ++i) {
            FileMetaData* f = files[i];
            Saver saver;
            saver.state = kNotFound;
            saver.ucmp = ucmp;
            saver.user_key = user_key;
            saver.value = value;
            // 通过table_cache_查询键
            s = vset_->table_cache_->Get(options, f->number, f->file_size, ikey,
                                    &saver, SaveValue);
            // 检查状态
            switch (saver.state) {
                case kNotFound:
                    break;  // 没有找到键的话，继续搜索下一个
                case kFound:
                    return s;  // 找到的话，就可以返回了
                case kDeleted:
                    s = Status::NotFound(Slice());  // 表示键被删除了，也就是可以确定这个键不存在，返回
                    return s;
                case kCorrupt:
                    s = Status::Corruption("corrupted key for ", user_key);  // 错误也返回
                    return s;
            }
        }
    }
    return Status::NotFound(Slice());  // 搜索完所有的键都没找到就返回。
}

// db/table_cache.cc
Status TableCache::Get(const ReadOptions& options, uint64_t file_number,
                        uint64_t file_size, const Slice& k, void* arg,
                        void (*handle_result)(void*, const Slice&,
                                                const Slice&)) {
    Cache::Handle* handle = nullptr;
    // 从表缓存里找到相应的file_number对应的文件
    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, handle_result);
    }
    return s;
}

// table/table.cc
Status Table::InternalGet(const ReadOptions& options, const Slice& k, void* arg,
                            void (*handle_result)(void*, const Slice&,
                                                const Slice&)) {
    Status s;
    // 搜索索引
    Iterator* iiter = rep_->index_block->NewIterator(rep_->options.comparator);
    iiter->Seek(k);
    // 找到一个索引项
    if (iiter->Valid()) {
        Slice handle_value = iiter->value();
        FilterBlockReader* filter = rep_->filter;
        BlockHandle handle;
        // 如果使用了布隆过滤器，则先查找布隆过滤器，如果没有发现，就直接返回了
        if (filter != nullptr && handle.DecodeFrom(&handle_value).ok() && !filter->KeyMayMatch(handle.offset(), k)) {
            // Not found
        } else {
            // 读取一个Data Block
            Iterator* block_iter = BlockReader(this, options, iiter->value());
            block_iter->Seek(k);
            if (block_iter->Valid()) {
                // 找到对应的键，调用回调函数
                (*handle_result)(arg, block_iter->key(), block_iter->value());
            }
            s = block_iter->status();
            delete block_iter;
        }
    }
    if (s.ok()) {
        s = iiter->status();
    }
    delete iiter;
    return s;
}

// db/version_set.cc
static void SaveValue(void* arg, const Slice& ikey, const Slice& v) {
    Saver* s = reinterpret_cast<Saver*>(arg);
    ParsedInternalKey parsed_key;
    if (!ParseInternalKey(ikey, &parsed_key)) {
        // 如果无法解析键，表示数据损坏
        s->state = kCorrupt;
    } else {
        // 比较对应的User Key是否相同
        if (s->ucmp->Compare(parsed_key.user_key, s->user_key) == 0) {
            s->state = (parsed_key.type == kTypeValue) ? kFound : kDeleted;
            if (s->state == kFound) {
                s->value->assign(v.data(), v.size());
            }
        }
    }
}

include/leveldb/db.h db/db_iter.h db/db_iter.cc: 实现了GET/PUT/DELETE
db/version_set.h db/version_set.cc: 实现从一个version的SSTable里读取一个键
db/memtable.h db/memtable.cc: 实现从MemTable读取一个键
inlclude/leveldb/table.h db/table_cache.h db/table_cache.cc db/table.cc: 定义从一个SSTable读取一个键

// db/version_edit.h
struct FileMetaData {
    FileMetaData() : refs(0), allowed_seeks(1 << 30), file_size(0) {}
    int refs;
    int allowed_seeks;     // Compaction时会介绍这个字段
    uint64_t number;
    uint64_t file_size;    // 文件大小
    InternalKey smallest;  // 文件里最小的internal key
    InternalKey largest;   // 文件里最大的internal key
};

class Version {
    VersionSet* vset_;  // 版本集的引用
    Version* next_;     // 版本在版本集里的next_指针
    Version* prev_;     // 版本在版本集里的prev_指针
    int refs_;          // 引用计数
    // SSTable的信息，每一项代表相应Level的SSTable信息
    // 除了Level 0外，每个Level里的文件都是按照最小键的顺序排列的，并且没有重叠
    // 通过这个数据项，搜索SSTable时，就可以从Level 0开始搜索
    std::vector<FileMetaData*> files_[config::kNumLevels];
}

// db/version_set.h
class VersionSet {
    Env* const env_;                      // 封装部分操作系统调用，包括文件、线程操作等
    const std::string dbname_;            // 数据库名称，Open时传入
    const Options* const options_;        // 数据库选项，Open时传入
    TableCache* const table_cache_;       // 打开的SSTable的缓存，Open时创建
    const InternalKeyComparator icmp_;    // 根据User Key生成的Internal Key的Comparator
    uint64_t next_file_number_;           // ldb、log和MANIFEST生成新文件时都有一个序号单调递增
    uint64_t manifest_file_number_;       // 当前的MANIFEST的编号
    uint64_t last_sequence_;              // 上一个使用的SequenceNumber
    uint64_t log_number_;                 // 当前的日志的编号
    WritableFile* descriptor_file_;       // MANIFEST打开的文件描述符
    log::Writer* descriptor_log_;         // MANIFEST实际存储的格式是WAL日志的格式，所以这里用来写入数据
    Version dummy_versions_;              // Version链表的头结点
    Version* current_;                    // 当前的Version
    // 这是用来记录Compact的进度，Compact总是从某一Level的最小的键开始到某个键结束，
    // 下次再从下一个键开始，所以这个就是下一次这个Level从哪个键开始Compact
    std::string compact_pointer_[config::kNumLevels];
}

// db/version_edit.h
class VersionEdit {
    typedef std::set<std::pair<int, uint64_t>> DeletedFileSet;
    std::string comparator_;             // 比较器的名称，持久化后，下次打开时需要对比一致
    uint64_t log_number_;                // 日志文件的编号
    uint64_t next_file_number_;          // ldb、log和MANIFEST下一个文件的编号
    SequenceNumber last_sequence_;       // 上一个使用的SequenceNumber
    bool has_comparator_;                // 记录上面4个字段是否存在，存在才会持久化的MANIFEST中
    bool has_log_number_;
    bool has_next_file_number_;
    bool has_last_sequence_
    // 和VersionSet里面的compact_pointers_相同
    std::vector<std::pair<int, InternalKey>> compact_pointers_;
    // 有哪些文件被删除，就是Version里哪些SSTable被删除
    DeletedFileSet deleted_files_;
    // 有哪些文件被增加，pair的第一个参数是Level，第二个参数是文件的元信息
    std::vector<std::pair<int, FileMetaData>> new_files_;
};

// db/version_edit.cc
enum Tag {
    kComparator = 1,            // 记录Comparator的名字
    kLogNumber = 2,             // 记录当前时刻的log_number
    kNextFileNumber = 3,        // 记录当前时刻的next_file_number_
    kLastSequence = 4,          // 记录当前时刻的last_sequence
    kCompactPointer = 5,        // 记录compact_pointer
    kDeletedFile = 6,           // 记录删除的文件信息
    kNewFile = 7                // 记录新增的文件信息
};

VersionSet::Builder builder(&vset, curr_version);  // 创建一个builder
builder.Apply(version_edit1);                      // 应用VersionEdit，可应用多个
builder.Apply(version_edit2);
Version new_version;               
builder.SaveTo(&new_version);                      // 将之前的Version和VersionEdit生成一个新的Version
vset.AppendVersion(new_version);                   // 将新Version添加到VersionSet

// db/version_set.cc
class VersionSet::Builder {
    typedef std::set<FileMetaData*, BySmallestKey> FileSet;
    // 表示某一Level，删除的文件和增加的文件
    struct LevelState {
        std::set<uint64_t> deleted_files;
        FileSet* added_files;
    };
    VersionSet* vset_;    // 对应的VersionSet
    Version* base_;       // 变化前的Version，均作为参数传入
    // 表示每一Level删除的文件和增加的文件，VersionEdit Apply时都累积到这里
    LevelState levels_[config::kNumLevels];
}

// db/version_set.cc
void Builder::Apply(VersionEdit* edit) {
    // 首先更新comact_pointer_，这个直接在VersionSet里更新，因为这个的信息只有一份
    // 新版本肯定是覆盖旧版本的，所以直接更新即可
    for (size_t i = 0; i < edit->compact_pointers_.size(); i++) {
        const int level = edit->compact_pointers_[i].first;
        vset_->compact_pointer_[level] = edit->compact_pointers_[i].second.Encode().ToString();
    }
    // 把VersionEdit里删除的文件插入到levels_相应Level里面去
    for (const auto& deleted_file_set_kvp : edit->deleted_files_) {
        const int level = deleted_file_set_kvp.first;
        const uint64_t number = deleted_file_set_kvp.second;
        levels_[level].deleted_files.insert(number);
    }
    // 把VersionEdit里添加的文件插入到levels_相应的Level里去
    for (size_t i = 0; i < edit->new_files_.size(); i++) {
        const int level = edit->new_files_[i].first;
        // 因为是新文件，构造一个FileMetaData
        FileMetaData* f = new FileMetaData(edit->new_files_[i].second);
        f->refs = 1;
        levels_[level].deleted_files.erase(f->number);
        levels_[level].added_files->insert(f);
    }
}

// db/version_set.cc
// 尝试将一个文件插入到新版本
void Builder::MaybeAddFile(Version* v, int level, FileMetaData* f) {
    if (levels_[level].deleted_files.count(f->number) > 0) {
        // 文件已删除
    } else {
        // 插入到Version的files里面
        std::vector<FileMetaData*>* files = &v->files_[level];
        f->refs++;
        files->push_back(f);
    }
}

void Builder::SaveTo(Version* v) {
    BySmallestKey cmp;
    cmp.internal_comparator = &vset_->icmp_;
    for (int level = 0; level < config::kNumLevels; level++) {
        // 拿出原本Version里的文件，以及Builder里累积的，添加的文件
        const std::vector<FileMetaData*>& base_files = base_->files_[level];
        std::vector<FileMetaData*>::const_iterator base_iter = base_files.begin();
        std::vector<FileMetaData*>::const_iterator base_end = base_files.end();
        const FileSet* added_files = levels_[level].added_files;
        v->files_[level].reserve(base_files.size() + added_files->size());
        // 按顺序进行合并
        for (const auto& added_file : *added_files) {
            // 找到base里面比added_file小的文件，添加到新的Version里
            // 采用MaybeAddFile，让被删除的文件无法添加
            for (std::vector<FileMetaData*>::const_iterator bpos =
                    std::upper_bound(base_iter, base_end, added_file, cmp);
                base_iter != bpos; ++base_iter) {
                MaybeAddFile(v, level, *base_iter);
            }
            MaybeAddFile(v, level, added_file);
        }
        // 添加剩下的文件
        for (; base_iter != base_end; ++base_iter) {
            MaybeAddFile(v, level, *base_iter);
        }
    }
}

// db/version_set.cc
// 输入参数edit表示的是改变的内容，比如一次Compaction可以得到edit
Status VersionSet::LogAndApply(VersionEdit* edit, port::Mutex* mu) {
    // 设定当前的log number
    if (edit->has_log_number_) {
    } else {
        edit->SetLogNumber(log_number_);
    }
    // 设定当前的next_file_number和last_sequence，这些都会被持久化到MANIFEST
    edit->SetNextFile(next_file_number_);
    edit->SetLastSequence(last_sequence_);
    // 创建一个新版本，新版本是current_和edit的结合
    Version* v = new Version(this);
    {
        Builder builder(this, current_);
        builder.Apply(edit);
        builder.SaveTo(v);
    }
    Finalize(v);
    std::string new_manifest_file;
    Status s;
    // 这里只有Open数据库的时候才会走到，如果需要保存新的MANIFEST，此时这个变量为null
    // 会创建一个新的MANIFEST，然后将当前的状态写入
    if (descriptor_log_ == nullptr) {
        new_manifest_file = DescriptorFileName(dbname_, manifest_file_number_);
        edit->SetNextFile(next_file_number_);
        s = env_->NewWritableFile(new_manifest_file, &descriptor_file_);
        if (s.ok()) {
            descriptor_log_ = new log::Writer(descriptor_file_);
            s = WriteSnapshot(descriptor_log_);
        }
    }
    {
        // 写文件时释放锁
        mu->Unlock();
        std::string record;
        edit->EncodeTo(&record);
        s = descriptor_log_->AddRecord(record);
        ...
        mu->Lock();
    }
    // 安装新版本，会把v放到VersionSet的链表中，然后将当前Version指向v
    AppendVersion(v);
    ...
    return s;
}

version_edit.h version_edit.cc: 实现VersionEdit
version_set.h version_set.cc：实现Version、VersionSet和Builder

void DBImpl::TEST_CompactRange(int level, const Slice* begin, const Slice* end)

// db/db_impl.h
struct ManualCompaction {
    int level;
    bool done;                 // 表示这次Manual Compaction是否完成
    const InternalKey* begin;  // 开始键，null表示最小键
    const InternalKey* end;    // 结束键，null表示最大键
    InternalKey tmp_storage;   // 保存Compaction的进度
};

// db/db_impl.cc
void DBImpl::TEST_CompactRange(int level, const Slice* begin,
                                const Slice* end) {
    // 构造ManualCompaction实例manual，保存Manual Compaction的信息
    MutexLock l(&mutex_);
    while (!manual.done) {
        // 如果manual.done为false，一直循环
        if (manual_compaction_ == nullptr) {
            // manual_compaction_是DBImpl的一个字段，后台线程会去检查这个字段，如果不为空，会触发Compaction
            // 所以将manual_compaction_赋值manual，后台线程就会检查触发Compaction
            manual_compaction_ = &manual;
            // 可能调度一次Compaction
            MaybeScheduleCompaction();
            // 设置完成后，会重新下一次循环，然后等待本次Compaction完成
        } else {
            // manual_compaction_已经被设置，表示已经有Manual Compaction进行中了
            // 等待后台线程完成，完成后，会重新执行循环
            background_work_finished_signal_.Wait();
        }
    }
    ...
}

// db/version_set.cc
void VersionSet::Finalize(Version* v) {
    int best_level = -1;     // 最大的比率的Level
    double best_score = -1;  // 最大的比率
    for (int level = 0; level < config::kNumLevels - 1; level++) {
        double score;
        if (level == 0) {
            // Level 0特殊处理，使用文件的个数，而不是大小来确定比率，因为对于大的writer-buffer
            // Level 0的文件会更大，这时候如果限定总大小，Compaction会偏多
            // 对于小的Level 0文件，数量会太多，影响读取的速度
            score = v->files_[level].size() /
                    static_cast<double>(config::kL0_CompactionTrigger);
        } else {
            // 计算文件总大小相对于最大大小的比率 
            const uint64_t level_bytes = TotalFileSize(v->files_[level]);
            score = static_cast<double>(level_bytes) / MaxBytesForLevel(options_, level);
        }
        if (score > best_score) {
            best_level = level;
            best_score = score;
        }
    }
    // 将计算得到的最大的score和level赋值，后台线程看到赋值后会开始Compaction
    v->compaction_level_ = best_level;
    v->compaction_score_ = best_score;
}

std::string VersionSet::compact_pointer_[config::kNumLevels];

static_cast<int>((f->file_size / 16384U))

// db/version_set.h
class Compaction {
    ...
    int level_;                      // Compaction文件所在的Level
    uint64_t max_output_file_size_;  // 生成的文件的最大值
    Version* input_version_;         // Compaction发生时的Version
    VersionEdit edit_;               // Compaction结果保存的VersionEdit
    std::vector<FileMetaData*> inputs_[2];
}

// db/db_impl.cc
void DBImpl::BackgroundCompaction() {
    ...
    Compaction* c;
    bool is_manual = (manual_compaction_ != nullptr);
    InternalKey manual_end;
    if (is_manual) {
        // 先处理Manual Compaction
        ManualCompaction* m = manual_compaction_;
        // 这里根据manual_compaction_的信息构造一个Compaction实例，表示需要完成的Compaction任务
        c = versions_->CompactRange(m->level, m->begin, m->end);
        // c为nullptr表示完成了
        m->done = (c == nullptr);
        if (c != nullptr) {
            // manual_end赋值为当前Compaction范围的结尾
            // 因为需要Compaction一个Level，防止一次Compaction太多数据，需要从最小键开始分段进行Compaction
            manual_end = c->input(0, c->num_input_files(0) - 1)->largest;
        }
    } else {
        c = versions_->PickCompaction();
    }
    ...
}

Compaction* VersionSet::PickCompaction() {
    Compaction* c;
    int level;
    const bool size_compaction = (current_->compaction_score_ >= 1);
    const bool seek_compaction = (current_->file_to_compact_ != nullptr);
    if (size_compaction) {
        // 优先考虑Size Compaction
        level = current_->compaction_level_;
        c = new Compaction(options_, level);
        // 根据compact_pointer_[level]找到下一个Compaction的文件f
        c->inputs_[0].push_back(f);
    } else if (seek_compaction) {
        // 对于Seek Compaction，文件已经确定了
        level = current_->file_to_compact_level_;
        c = new Compaction(options_, level);
        c->inputs_[0].push_back(current_->file_to_compact_);
    } else {
        return nullptr;
    }
    ..
    // 如果c->inputs_里的文件是Level 0，那么同一Level其它文件也有可能有重叠，则找到其它重叠的文件
    // 找到上一层Level和选定的Level有重叠的文件，这样就找到了两层需要Compaction的文件
    SetupOtherInputs(c);
    return c;
}

// db/db_impl.cc
void DBImpl::BackgroundCompaction() {
    ...
    if (c == nullptr) {
        // 表示没有需要Compaction的内容
    } else if (!is_manual && c->IsTrivialMove()) {
        // 处理一种特殊情况，也就是参与Compaction的文件，level_有一个文件，而level_ + 1 没有
        // 这时候只需要直接更改元数据，然后文件移动到level_ + 1即可，不需要多路归并
    } else {
        // CompactionState保存Compaction的状态
        CompactionState* compact = new CompactionState(c);
        // 实际的Compaction
        status = DoCompactionWork(compact);
        // 清理操作
    }
    ...
}

// db/db_impl.cc
Status DBImpl::DoCompactionWork(CompactionState* compact) {
    // 取当前最小的使用中的SequenceNumber
    if (snapshots_.empty()) {
        compact->smallest_snapshot = versions_->LastSequence();
    } else {
        compact->smallest_snapshot = snapshots_.oldest()->sequence_number();
    }
    // 参与Compaction的SSTable组成一个迭代器
    Iterator* input = versions_->MakeInputIterator(compact->compaction);
    input->SeekToFirst();
    ParsedInternalKey ikey;
    std::string current_user_key;       // 记录当前的User Key
    bool has_current_user_key = false;  // 记录是否碰到过一个同样的User Key
    SequenceNumber last_sequence_for_key = kMaxSequenceNumber;
    while (input->Valid()) {
    ...
        Slice key = input->key();
        if (compact->compaction->ShouldStopBefore(key) && compact->builder != nullptr) {
            // ShouldStopBefore判断生成的SSTable和level_ + 2层的有重叠的文件个数，如果超过10个，
            // 那么这个SSTable生成就完成了,这样保证了新生产的SSTable和上一层不会有过多的重叠
            // 创建一个新的SSTable，写入文件
            status = FinishCompactionOutputFile(compact, input);
        }
        bool drop = false;
        if (!has_current_user_key || user_comparator()->Compare(ikey.user_key, Slice(current_user_key)) != 0) {
            // 如果第一次碰到一个User Key
            current_user_key.assign(ikey.user_key.data(), ikey.user_key.size());
            has_current_user_key = true;
            last_sequence_for_key = kMaxSequenceNumber;
        }
        // 如果上一个Key的SequenceNumber <= 最小的存活的Snapshot，那么
        // 这个Key的SequenceNumber一定 < 最小的存活的Snapshot，那么这个Key就不
        // 会被任何线程看到了，可以被丢弃，上面碰到了第一个User Key时，设置了
        // last_sequence_for_key = kMaxSequenceNumber;保证第一个Key一定不会
        // 被丢弃。
        if (last_sequence_for_key <= compact->smallest_snapshot) {
            drop = true;  // (A)
        } else if (ikey.type == kTypeDeletion &&
                    ikey.sequence <= compact->smallest_snapshot &&
                    compact->compaction->IsBaseLevelForKey(ikey.user_key))  {
            // 如果碰到了一个删除操作，并且SequenceNumber <= 最小的Snapshot，
            // 通过IsBaseLevelForKey判断更高Level不会有这个User Key存在，那么这个Key就被丢弃
            drop = true;
        }
        last_sequence_for_key = ikey.sequence;
        if (!drop) {
            ...
            // 没有被丢弃就添加Key
            compact->builder->Add(key, input->value());
            // 达到文件大小，就写入文件，生成新文件
            if (compact->builder->FileSize() >=
                    compact->compaction->MaxOutputFileSize()) {
                status = FinishCompactionOutputFile(compact, input);
            }
        }
        input->Next();
    }
    ...
    // 安装Compaction的结果
    status = InstallCompactionResults(compact);
    ...
}

// db/db_impl.cc
Status DBImpl::InstallCompactionResults(CompactionState* compact) {
    ...
    // 将改删除的文件和改添加的文件更新到VersionEdit里
    compact->compaction->AddInputDeletions(compact->compaction->edit());
    const int level = compact->compaction->level();
    for (size_t i = 0; i < compact->outputs.size(); i++) {
        const CompactionState::Output& out = compact->outputs[i];
        compact->compaction->edit()->AddFile(level + 1, out.number, out.file_size,
                                            out.smallest, out.largest);
    }
    // 应用一次版本变更，安装新版本
    return versions_->LogAndApply(compact->compaction->edit(), &mutex_);
}

Version_set.h version_set.cc: Compaction类的实现，更新Compaction的统计信息，确定Compaction的内容
db_impl.h db_impl.cc: Compaction实际操作的实现

Status DB::Open(const Options& options, const std::string& dbname, DB** dbptr)

// db/db_impl.cc

Status DB::Open(const Options& options, const std::string& dbname, DB** dbptr) {
    *dbptr = nullptr;

    DBImpl* impl = new DBImpl(options, dbname);
    VersionEdit edit;
    bool save_manifest = false;
    // 调用DBImpl::Recover完成MANIFEST的加载和故障恢复
    Status s = impl->Recover(&edit, &save_manifest);
    if (s.ok() && impl->mem_ == nullptr) {
        // 创建日志和相应的MemTable
    }
    if (s.ok() && save_manifest) {
        // 如果需要重写MANIFEST文件，那么做一个版本变更，这里面会创建一个新的MANIFEST
        // 将当前的版本信息写入，然后将edit的内容写入。
        edit.SetLogNumber(impl->logfile_number_);
        s = impl->versions_->LogAndApply(&edit, &impl->mutex_);
    }

    ...

    if (s.ok()) {
        *dbptr = impl;
    } else {
        delete impl;
    }
    return s;
}

// db/db_impl.cc
Status DBImpl::Recover(VersionEdit* edit, bool* save_manifest) {
    env_->CreateDir(dbname_);  // 创建数据库目录
    // 加文件锁，防止其他进程进入
    Status s = env_->LockFile(LockFileName(dbname_), &db_lock_);
    if (!env_->FileExists(CurrentFileName(dbname_))) {
        // 如果CURRENT文件不存在，说明需要新创建数据库
        s = NewDB();
    }
    // 读取MANIFEST文件进行版本信息的恢复
    s = versions_->Recover(save_manifest);
    // 之前的MANIFEST恢复，会得到版本信息，里面包含了之前的log number
    // 搜索文件系统里的log，如果这些日志的编号 >= 这个log number，那么这些
    // 日志都是关闭时丢失的数据，需要恢复，这里将日志按顺序存储在logs里面
    // 逐个恢复日志的内容
    for (size_t i = 0; i < logs.size(); i++) {
        s = RecoverLogFile(logs[i], (i == logs.size() - 1), save_manifest, edit, &max_sequence);
        ...
    }
    ...
    return Status::OK();
}

// db/db_impl.cc
Status DBImpl::NewDB() {
    VersionEdit new_db;
    // 保存比较器的名称，下次打开时需要用相同的名称打开
    new_db.SetComparatorName(user_comparator()->Name());  
    new_db.SetLogNumber(0);  // 分配日志文件的编号为0                      
    new_db.SetNextFile(2);   // 下一个待分配的文件编号是2，因为1分配给了MANIFEST文件
    new_db.SetLastSequence(0);
    // 创建MANIFEST文件，将VersionEdit写入
    const std::string manifest = DescriptorFileName(dbname_, 1);
    WritableFile* file;
    Status s = env_->NewWritableFile(manifest, &file);
    log::Writer log(file);
    std::string record;
    new_db.EncodeTo(&record);
    s = log.AddRecord(record);
    // 让CURRENT文件指向这个MANIFEST文件
    s = SetCurrentFile(env_, dbname_, 1);
    ...
    return s;
}

// db/version_set.cc
Status VersionSet::Recover(bool* save_manifest) {
    // 读取CURRENT文件的内容，获取当前使用的MANIFEST文件
    // 读取MANIFEST文件，将里面的VersionEdit读取应用到一个builder里
    Version* v = new Version(this);      // 创建一个新版本版本
    builder.SaveTo(v);                   // 将builder里面的内容应用到新版本里
    Finalize(v);                         // 更新Size Compaction的统计信息
    AppendVersion(v);                    // 安装新版本成为当前版本
    if (ReuseManifest(dscname, current)) {
        // 如果MANIFEST可以重用，那么不需要保存MANIFEST
        // 这里主要判断MANIFEST的大小，如果大于2M，那么就不会重用MANIFEST文件，
        // 而是将当前状态写入到一个新的MANIFEST文件里，这样可以避免打开的时候读取
        // 太大的MANINFEST，使得打开时间太长
    } else {
        *save_manifest = true;
    }
    return s;
}

db/db_impl.h db/db_impl.cc: `DB:Open`的入口
db/version_set.h db/version_set.cc: 完成MANIFEST文件的读取和构造版本信息