LSM-Tree - LevelDb 源码解析（二）

struct FileMetaData {
  FileMetaData() : refs(0), allowed_seeks(1 << 30), file_size(0) {}
  int refs;
  int allowed_seeks; // 允许压缩搜索
  uint64_t number;
  uint64_t file_size; // 文件大小
  InternalKey smallest; // 表提供的最小内部密钥
  InternalKey largest; // 表提供最大内部密钥
};

void DBImpl::CompactMemTable() {
  VersionEdit edit;
  Version* base = versions_->current();
  WriteLevel0Table(imm_, &edit, base);
  versions_->LogAndApply(&edit, &mutex_);
  RemoveObsoleteFiles();
}

Status DBImpl::WriteLevel0Table(MemTable* mem, VersionEdit* edit,
Version* base) {
  // SSTable文件信息
  FileMetaData meta;
  meta.number = versions_->NewFileNumber();
  pending_outputs_.insert(meta.number);
  Iterator* iter = mem->NewIterator();
  // 构建SSTable文件
  BuildTable(dbname_, env_, options_, table_cache_, iter, &meta);
  pending_outputs_.erase(meta.number);
  // 注意，如果 file_size 为零，则该文件已被删除，并且不应被添加到清单中。
  // 获取文件信息的最大值和最小值
  const Slice min_user_key = meta.smallest.user_key();
  const Slice max_user_key = meta.largest.user_key();
  // level层级扫描
  base->PickLevelForMemTableOutput(min_user_key, max_user_key);
  // 写入文件
  edit->AddFile(level, meta.number, meta.file_size, meta.smallest,
  return Status::ok();

void DBImpl::BackgroundCompaction() {  
  // 如果存在不可变imumem,进行压缩合并
  CompactMemTable();
  versions_->PickCompaction();
  VersionSet::LevelSummaryStorage tmp;
  CompactionState* compact = new CompactionState(c);
  DoCompactionWork(compact);
  CleanupCompaction(compact);
  c->ReleaseInputs();
  RemoveObsoleteFiles();
}

Status DBImpl::DoCompactionWork(CompactionState* compact) {
  int64_t imm_micros = 0; // Micros spent doing imm_ compactions
  if (snapshots_.empty()) {
    // 快照为空，找到直接采用记录信息的最后序列号
    compact->smallest_snapshot = versions_->LastSequence();
  } else {
    // 快照存在，则抛弃之前所有的序列
    compact->smallest_snapshot = snapshots_.oldest()->sequence_number();
  }
  // 对于待压缩数据进行，内部生成一个MergingIterator，当构建迭代器之后键内部就是有序的状态了，也就是前面说的归并排序的部分
  Iterator* input = versions_->MakeInputIterator(compact->compaction);
  input->SeekToFirst();
  Status status;
  ParsedInternalKey ikey;
  std::string current_user_key;
  //当前记录user key
  bool has_current_user_key = false;
  SequenceNumber last_sequence_for_key = kMaxSequenceNumber;
  while (input->Valid() && !shutting_down_.load(std::memory_order_acquire)) {
  // 优先考虑imumemtable的压缩工作
  if (has_imm_.load(std::memory_order_relaxed)) {
    const uint64_t imm_start = env_->NowMicros();
    imm_micros += (env_->NowMicros() - imm_start);
  }
  Slice key = input->key();
  if (compact->compaction->ShouldStopBefore(key) &&
    compact->builder != nullptr) {
    status = FinishCompactionOutputFile(compact, input);
  }
  // 处理键/值，添加到状态等。
  bool drop = false;
  if (!ParseInternalKey(key, &ikey)) {
    // 删除和隐藏呗删除key
    current_user_key.clear();
    has_current_user_key = false;
    // 更新序列号
    last_sequence_for_key = kMaxSequenceNumber;
  } else {
    if (!has_current_user_key ||
      user_comparator()->Compare(ikey.user_key, Slice(current_user_key)) !=
    0) {
    // 用户key第一次出现
    current_user_key.assign(ikey.user_key.data(), ikey.user_key.size());
    has_current_user_key = true;
    last_sequence_for_key = kMaxSequenceNumber;
  }
  if (last_sequence_for_key <= compact->smallest_snapshot) {
    // 压缩以后旧key边界的被新的覆盖
    drop = true; // (A)
  } else if (ikey.type == kTypeDeletion &&
    ikey.sequence <= compact->smallest_snapshot &&
    compact->compaction->IsBaseLevelForKey(ikey.user_key)) {
    // 对于这个用户密钥：
    // (1) 高层没有数据
    // (2) 较低层的数据会有较大的序列号
    // (3) 层中的数据在此处被压缩并具有
    // 较小的序列号将在下一个被丢弃
    // 这个循环的几次迭代（根据上面的规则（A））。
    // 因此，此删除标记已过时，可以删除。
    drop = true;
  }
  last_sequence_for_key = ikey.sequence;
}
}

CompactionStats stats;
  stats.micros = env_->NowMicros() - start_micros - imm_micros;
  //选择两个层级的SSTable
  for (int which = 0; which < 2; which++) {
    for (int i = 0; i < compact->compaction->num_input_files(which); i++) {
      stats.bytes_read += compact->compaction->input(which, i)->file_size;
    }
  }
  for (size_t i = 0; i < compact->outputs.size(); i++) {
    stats.bytes_written += compact->outputs[i].file_size;
  }
  // 压缩到更高的层级
  stats_[compact->compaction->level() + 1].Add(stats);
  // 注册压缩结果
  InstallCompactionResults(compact);
  // 压缩信息存储
  VersionSet::LevelSummaryStorage tmp;
  Log(options_.info_log, "compacted to: %s", versions_->LevelSummary(&tmp));
  return status;

Status VersionSet::LogAndApply(VersionEdit* edit, port::Mutex* mu) {
  // 更新版本链表信息
  if (!edit->has_prev_log_number_) {
    edit->SetPrevLogNumber(prev_log_number_);
  }
  edit->SetNextFile(next_file_number_);
  edit->SetLastSequence(last_sequence_);
  Version* v = new Version(this);
  // 构建当前的版本version，委托给建造器进行构建
  Builder builder(this, current_);
  builder.Apply(edit);
  builder.SaveTo(v);
  // 关键方法：内部通过打分机制确定文件所在的层级，值得注意的是level0的层级确定在源代码中有较多描述
  Finalize(v);
// 如有必要，通过创建包含当前版本快照的临时文件来初始化新的描述符日志文件。
  std::string new_manifest_file;
  //  没有理由在这里解锁*mu，因为我们只在第一次调用LogAndApply时（打开数据库时）碰到这个路径。
  new_manifest_file = DescriptorFileName(dbname_, manifest_file_number_);
  // 写入mainfest文件
  env_->NewWritableFile(new_manifest_file, &descriptor_file_);
  // 写入版本信息快照
  WriteSnapshot(descriptor_log_);
  // 把记录写到 MANIFEST中
  descriptor_log_->AddRecord(record);
  //如果创建了新的文件，则将当前版本指向这个文件
  SetCurrentFile(env_, dbname_, manifest_file_number_);
  // 注册新版本
  AppendVersion(v);
  return Status::OK();
}

// TableBuilder 提供了用于构建 Table 的接口
//（从键到值的不可变且排序的映射）。
//
// 多个线程可以在 TableBuilder 上调用 const 方法，而无需
// 外部同步，但如果任何线程可能调用
// 非常量方法，所有访问同一个 TableBuilder 的线程都必须使用
// 外部同步。
class LEVELDB_EXPORT TableBuilder {
public:
  TableBuilder(const Options& options, WritableFile* file);
  TableBuilder(const TableBuilder&) = delete;
  TableBuilder& operator=(const TableBuilder&) = delete;
  /*
  改变该构建器所使用的选项。注意：只有部分的
  选项字段可以在构建后改变。如果一个字段是
  不允许动态变化，并且其在结构中的值
  中的值与传递给本方法的结构中的值不同。
  结构中的值不同，该方法将返回一个错误
  而不改变任何字段。
  */
  Status ChangeOptions(const Options& options);
  void Add(const Slice& key, const Slice& value);
  void Flush();
  Status status() const;
  Status Finish();
  /*
  表示应该放弃这个建设者的内容。停止
  在此函数返回后停止使用传递给构造函数的文件。
  如果调用者不打算调用Finish()，它必须在销毁此构建器之前调用Abandon()
  之前调用Abandon()。
  需要。Finish()、Abandon()未被调用。
  */
  void Abandon();
  uint64_t NumEntries() const;
  uint64_t FileSize() const;
  private:
    bool ok() const { return status().ok(); }
    void WriteBlock(BlockBuilder* block, BlockHandle* handle);
    void WriteRawBlock(const Slice& data, CompressionType, BlockHandle* handle);
    struct Rep;
    Rep* rep_;
  };
}