leveldb源码阅读5 Compaction And Version

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
void DBImpl::MaybeScheduleCompaction() {
  mutex_.AssertHeld();
  // 检查是否有后台压缩任务被调度，如果有就不做操作
  if (background_compaction_scheduled_) {
    // Already scheduled 
  } else if (shutting_down_.load(std::memory_order_acquire)) {
    // DB is being deleted; no more background compactions
    // DB正在关闭时，不能调度任何压缩任务
  } else if (!bg_error_.ok()) {
    // Already got an error; no more changes
    // 有错误时，不能进行任何压缩任务
  } else if (imm_ == nullptr && manual_compaction_ == nullptr &&
             !versions_->NeedsCompaction()) {
    // No work to be done
    // 没有任何待处理的压缩，则不调度压缩
  } else {
    // 当避免了各种不必要的操作和潜在问题时，可以调度压缩任务
    background_compaction_scheduled_ = true;
    env_->Schedule(&DBImpl::BGWork, this);
  }
}

1
2
3
4
5
6
7
// Returns true iff some level needs a compaction.
  bool NeedsCompaction() const {
    Version* v = current_;
     // compaction_score_字段用来计算是否需要触发Size Compaction，
     // file_to_compact_用来计算是否需要触发Seek Compaction
    return (v->compaction_score_ >= 1) || (v->file_to_compact_ != nullptr);
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
/*确立memtable输出的数据应该存放哪个层级*/
int Version::PickLevelForMemTableOutput(const Slice& smallest_user_key,
                                        const Slice& largest_user_key) {
  int level = 0;
  // 看level0是否与memtable输出的键有重叠，如果没有，继续向下查找
  // 没有重叠向下查找是为了确保多个层级之间都没有重叠，减少对于major的调用，提升性能
  // 存在重叠的愿意按：多版本的不同数据，同一个文件的多个用户间，文件分布不均等
  // major就是拿来处理层与层之间的重叠，进而定期优化清理层级结构。
  if (!OverlapInLevel(0, &smallest_user_key, &largest_user_key)) {
    // Push to next level if there is no overlap in next level,
    // and the #bytes overlapping in the level after that are limited.
    InternalKey start(smallest_user_key, kMaxSequenceNumber, kValueTypeForSeek);
    InternalKey limit(largest_user_key, 0, static_cast<ValueType>(0));
    std::vector<FileMetaData*> overlaps;
    // 循环遍历，直到达到最大内存压缩层级
    while (level < config::kMaxMemCompactLevel) {
      if (OverlapInLevel(level + 1, &smallest_user_key, &largest_user_key)) {
        break; // 有重叠时保持当层级退出
      }
      // 如下面两层的重叠字数超过最大值时，容易导致性能问题，层级不能上升
      if (level + 2 < config::kNumLevels) {
        // Check that file does not overlap too many grandparent bytes.
        GetOverlappingInputs(level + 2, &start, &limit, &overlaps);
        const int64_t sum = TotalFileSize(overlaps);
        if (sum > MaxGrandParentOverlapBytes(vset_->options_)) {
          break;
        }
      }
      level++;
    }
  }
  return level;
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
/*将db通过类型转换程DBImpl，然后调用BackGroundCall接口，启动和调度后台处理任务。*/
void DBImpl::BGWork(void* db) {
  reinterpret_cast<DBImpl*>(db)->BackgroundCall();
}

/*处理levelDB的后台任务*/
void DBImpl::BackgroundCall() {
  MutexLock l(&mutex_); // 创建互斥锁，保证线程安全。
  assert(background_compaction_scheduled_); // 确保后台有任务被调度
  if (shutting_down_.load(std::memory_order_acquire)) {
    // No more background work when shutting down.
    // 数据库正在关闭时不执行任何后台工作，避免关闭过程中继续处理数据导致数据损坏
  } else if (!bg_error_.ok()) {
    // No more background work after a background error.
    // 排除后台错误
  } else {
    // 执行合并操作，将levelDB中的数据整理并优化到适当的层级
    // 只要满足条件触发合并，这个过程就会不断被调用
    // 首次遍历，优化存储，减少重叠文件和荣誉数据
    BackgroundCompaction();
  }

  // 重置调度装置，表示任务完成，避免重复执行同一个合并任务
  background_compaction_scheduled_ = false;

  // Previous compaction may have produced too many files in a level,
  // so reschedule another compaction if needed.
  MaybeScheduleCompaction(); // 检查是否需要重新调度合并，对sstable文件数量进行评估
  background_work_finished_signal_.SignalAll(); // 唤醒等待线程，可以促进触发新的合并
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
/*leveldb的核心部分，涵盖了合并的选择，执行和状态管理*/
void DBImpl::BackgroundCompaction() {
    // 上锁保障线程安全
    mutex_.AssertHeld();  
    // 处理尚未合并的immutable memetable
    if (imm_ != nullptr) { 
        CompactMemTable();
        return;
}
    
    // 处理尚未合并的 munual compaction
    // 根据is_munual选择对c进行手动合并还是自动合并
    Compaction* c;
    bool is_manual = (manual_compaction_ != nullptr); 
    
    // 对c进行再次检查的主要原因是为了确保合并操作的完整性和正确性。
    // 第一次合并主要是处理选择出来的SSTable文件，执行合并操作。
    // 第二次目的是确认合并结果
    Status status;
    /*c为空时，跳过；
      当前任务不是mutual compaction，则判断是不是is_trivial move(只需要从一层移动到下一层)，既不需要合并也不需要拆分SSTable，只通过Versionset修改一下level的文件编号即可。
      否则，执行compaction的操作，依次执行DoCompactionWork、CleanupCompaction、RemoveObsoleteFiles。*/
    if (c == nullptr) {
    // Nothing to do 没有合并文件
    } else if (!is_manual && c->IsTrivialMove()) {...}
   
    // 错误处理，状态更新
}     
// is_manual为1则是手动合并，为0则是自动合并。

1
2
3
4
5
6
7
8
9
10
11
12
13
/*判断当前的操作是否可以简单移动*/
bool Compaction::IsTrivialMove() const {
  const VersionSet* vset = input_version_->vset_;
  // Avoid a move if there is lots of overlapping grandparent data.
  // Otherwise, the move could create a parent file that will require
  // a very expensive merge later on.
  // 当前层有文件而下一层没有文件，因为这确保了移动不会引入重叠，并简化了未来的合并流程
  // 如果在进行简单移动时，当前层的文件与祖父层的文件重叠过多，这可能导致在未来需要进行昂贵的合并操作
  // 如果当前层与下一层之间的重叠较少，则在合并时主要关注当前层和下一层之间的关系，而与祖父层的关系相对较小。也就是让祖父层和下一层重叠多是一个很糟的决定
  return (num_input_files(0) == 1 && num_input_files(1) == 0 &&
          TotalFileSize(grandparents_) <=
              MaxGrandParentOverlapBytes(vset->options_));
}

1
2
3
4
5
6
7
8
9
10
/*
leveldb是流控和等待的方式。
在这个函数中，通过综合内存管理和文件状态，确保leveldb在写入时的高效性。
在循环写入的过程中，会对不同的情况进行判断，比如
1.level0文件数量接近上限(默认为8，超过4时会触发minor compaction），开始对写入操作施加延迟，减少坚持波动给压缩线程留出时间
2.内存表有足够空间时，继续写入，memtable有空间
3.内存表已满，并且之前的内存表正在压缩（memtable已满，immutable memtable没有完成minor compaction），等待
4.磁盘中level0文件太多，超过8时，等待
5.切换到新的内存表，压缩之前的内存表，更新状态
*/

1
2
3
4
5
6
7
void DBImpl::BackgroundCall() {  
  // ... ...
  // Previous compaction may have produced too many files in a level,
  // so reschedule another compaction if needed.
  MaybeScheduleCompaction();
  background_work_finished_signal_.SignalAll();
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
Status DBImpl::Get(const ReadOptions& options, const Slice& key, std::string* value) {
  Status s;
  MutexLock l(&mutex_); // 加锁保护
  SequenceNumber snapshot;
  if (options.snapshot != nullptr) {
    snapshot = // 确认快照
        static_cast<const SnapshotImpl*>(options.snapshot)->sequence_number();
  } else {
    snapshot = versions_->LastSequence();
  }

  MemTable* mem = mem_; // 获取当前的内存表和不可变内存表
  MemTable* imm = imm_;
  Version* current = versions_->current(); // 当前版本的引用
  mem->Ref();
  if (imm != nullptr) 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 != nullptr && imm->Get(lkey, value, &s)) {
      // Done 再在不可变内存表中找
    } else {
      // 最后在当前版本的SSTable文件中查找
      s = current->Get(options, lkey, value, &stats);
      have_stat_update = true;
    }
    mutex_.Lock();
  }

  // 如果查找时更新了统计信息，调用MaybeScheduleCompaction决定是否需要进行压缩
  if (have_stat_update && current->UpdateStats(stats)) {
    MaybeScheduleCompaction(); // 这个会对各种压缩是否需要进行检查
  }
  // 解除引用计数
  mem->Unref();
  if (imm != nullptr) imm->Unref();
  current->Unref();
  return s;
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
Status Version::Get(const ReadOptions& options, const LookupKey& k,
                    std::string* value, GetStats* stats) {
  stats->seek_file = nullptr;
  stats->seek_file_level = -1;

  struct State {
    Saver saver;
    GetStats* stats;
    const ReadOptions* options;
    Slice ikey;
    FileMetaData* last_file_read;
    int last_file_read_level;

    VersionSet* vset;
    Status s;
    bool found;

    static bool Match(void* arg, int level, FileMetaData* f) {
      State* state = reinterpret_cast<State*>(arg);
      // 没有找到文件时，记录当前文件和层级
      if (state->stats->seek_file == nullptr &&
          state->last_file_read != nullptr) {
        // 检查是否有文件已经被访问过
        // We have had more than one seek for this read.  Charge the 1st file.
        state->stats->seek_file = state->last_file_read;
        state->stats->seek_file_level = state->last_file_read_level;
      }

      state->last_file_read = f;
      state->last_file_read_level = level;

      // 使用tablecache从缓存中获取数据
      state->s = state->vset->table_cache_->Get(*state->options, f->number,
                                                f->file_size, state->ikey,
                                                &state->saver, SaveValue);
      if (!state->s.ok()) {
        state->found = true;
        return false;
      }
      switch (state->saver.state) {
        case kNotFound: // 继续找其它文件
          return true;  // Keep searching in other files
        case kFound: // 找到键更新状态返回
          state->found = true;
          return false;
        case kDeleted: // 返回，键已删除
          return false;
        case kCorrupt: // 记录错误状态返回
          state->s =
              Status::Corruption("corrupted key for ", state->saver.user_key);
          state->found = true;
          return false;
      }

      // Not reached. Added to avoid false compilation warnings of
      // "control reaches end of non-void function".
      return false;
    }
  };

  State state;
  state.found = false;
  state.stats = stats;
  state.last_file_read = nullptr;
  state.last_file_read_level = -1;

  state.options = &options;
  state.ikey = k.internal_key();
  state.vset = vset_;

  state.saver.state = kNotFound;
  state.saver.ucmp = vset_->icmp_.user_comparator();
  state.saver.user_key = k.user_key();
  state.saver.value = value;
    
  /*遍历版本中的文件，查找与给定的用户键和内部键重叠的文件，并通过回调函数处理这些文件。通过分层的方式，确保在最新文件优先的同时，也不遗漏其他层中的重叠文件。*/
  ForEachOverlapping(state.saver.user_key, state.ikey, &state, &State::Match);

  return state.found ? state.s : Status::NotFound(Slice());
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
/*将imm变为SSTable，同时添加到leveldb的版本控制和文件管理。*/
void DBImpl::CompactMemTable() {
  mutex_.AssertHeld(); // 操作期间有锁，避免并发问题（并发控制）
  assert(imm_ != nullptr);

  // Save the contents of the memtable as a new Table
  VersionEdit edit; // 版本编辑（内存管理）
  Version* base = versions_->current(); 
  base->Ref(); // 引用版本
  Status s = WriteLevel0Table(imm_, &edit, base); // 将imm写入level0，更新版本信息
  base->Unref();
  // 如果写入成功并且正在关闭数据库，记录错误返回
  if (s.ok() && shutting_down_.load(std::memory_order_acquire)) {
    s = Status::IOError("Deleting DB during memtable compaction");
  }
  // 如果没有错误，更新版本信息（错误管理）
  // Replace immutable memtable with the generated Table
  if (s.ok()) {
    edit.SetPrevLogNumber(0);
    edit.SetLogNumber(logfile_number_);  // Earlier logs no longer needed
    s = versions_->LogAndApply(&edit, &mutex_);
  }
  // 清理空间，清理错误文件（性能优化）
  if (s.ok()) {
    // Commit to the new state
    imm_->Unref();
    imm_ = nullptr;
    has_imm_.store(false, std::memory_order_release);
    RemoveObsoleteFiles();
  } else {
    RecordBackgroundError(s);
  }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
/*要用于将内存表（MemTable）的数据写入到 Level0的SST文件中*/
/*
* meta文件元数据，通过mem的迭代器保存到sst文件中
* edit描述数据库版本的更改，记录新文件的添加和删除等信息
* base当前数据库版本，参考已经存在的数据和文件信息
*/
Status DBImpl::WriteLevel0Table(MemTable* mem, VersionEdit* edit,
                                Version* base) {
  mutex_.AssertHeld(); // 确保有锁
  const uint64_t start_micros = env_->NowMicros(); // 记录开始时间
  FileMetaData meta; // 创建文件元数据
  meta.number = versions_->NewFileNumber();
  pending_outputs_.insert(meta.number);
  Iterator* iter = mem->NewIterator(); // 创建迭代器遍历内存表
  Log(options_.info_log, "Level-0 table #%llu: started",
      (unsigned long long)meta.number); // 记录日志

  Status s;
  { // 加锁建构表
    mutex_.Unlock();  // （理清表中有什么，meta中有什么）
    /*
    BuildTable 函数负责实际的写入过程。
    它将遍历迭代器中的数据，并写入到新SST文件中。
    会在指定数据库目录中创建新文件，文件名基于meta.number生成。
    */
    s = BuildTable(dbname_, env_, options_, table_cache_, iter, &meta);
    mutex_.Lock();
  }
  // 日志记录和删除迭代器 
  Log(options_.info_log, "Level-0 table #%llu: %lld bytes %s",
      (unsigned long long)meta.number, (unsigned long long)meta.file_size,
      s.ToString().c_str());
  delete iter;
  pending_outputs_.erase(meta.number); // 这个pending_outputs什么作用？
  // 主要用于跟踪正在进行的写入操作，避免同一时刻对同一个文件进行多次写入
  // 防止并发造成的数据破坏

  // Note that if file_size is zero, the file has been deleted and
  // should not be added to the manifest.
  int level = 0; // 表格构建成功，文件大小大于0
  if (s.ok() && meta.file_size > 0) {
    const Slice min_user_key = meta.smallest.user_key();
    const Slice max_user_key = meta.largest.user_key();
    if (base != nullptr) {
      // 根据用户键选择写入层级
      level = base->PickLevelForMemTableOutput(min_user_key, max_user_key);
    }
    // 将文件信息添加到版本编辑中
    edit->AddFile(level, meta.number, meta.file_size, meta.smallest,
                  meta.largest);
  }

  CompactionStats stats;
  stats.micros = env_->NowMicros() - start_micros;
  stats.bytes_written = meta.file_size;
  stats_[level].Add(stats);
  return s;
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
/*处理leveldb中的范围压缩*/
void DBImpl::CompactRange(const Slice* begin, const Slice* end) {
  int max_level_with_files = 1;
  {
    MutexLock l(&mutex_);
    Version* base = versions_->current(); // 获取当前版本
    for (int level = 1; level < config::kNumLevels; level++) {
      if (base->OverlapInLevel(level, begin, end)) {
        max_level_with_files = level; // 查找有重叠的最高层级
      }
    }
  }
  TEST_CompactMemTable();  // TODO(sanjay): Skip if memtable does not overlap
  for (int level = 0; level < max_level_with_files; level++) {
    TEST_CompactRange(level, begin, end); // 对每个层级进行范围压缩
    // 将指定范围的sst文件合并，更新数据库存储结构
  }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
/*执行具体的压缩工作，从输入迭代器中读取键值对，合并数据，写入输出文件，更新版本信息*/
Status DBImpl::DoCompactionWork(CompactionState* compact) {
  const uint64_t start_micros = env_->NowMicros();// 初始化开始时间
  int64_t imm_micros = 0;  // Micros spent doing imm_ compactions
  // 记录日志
  Log(options_.info_log, "Compacting %d@%d + %d@%d files",
      compact->compaction->num_input_files(0), compact->compaction->level(),
      compact->compaction->num_input_files(1),
      compact->compaction->level() + 1);

  assert(versions_->NumLevelFiles(compact->compaction->level()) > 0);
  assert(compact->builder == nullptr);
  assert(compact->outfile == nullptr);
  if (snapshots_.empty()) {// 设置快照
    compact->smallest_snapshot = versions_->LastSequence();
  } else {
    compact->smallest_snapshot = snapshots_.oldest()->sequence_number();
  }
  // 为当前版本创建一个迭代器，用于遍历待压缩的键值对
  Iterator* input = versions_->MakeInputIterator(compact->compaction);

  // Release mutex while we're actually doing the compaction work
  mutex_.Unlock();

  input->SeekToFirst(); // 检查迭代器是否有效
  Status status;
  ParsedInternalKey ikey;
  std::string current_user_key;
  bool has_current_user_key = false;
  SequenceNumber last_sequence_for_key = kMaxSequenceNumber;

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
  /*遍历处理键值对，根据压缩的需求决定哪些键值对保留，哪些需要抛弃*/
  while (input->Valid() && !shutting_down_.load(std::memory_order_acquire)) {
    // Prioritize immutable compaction work优先处理imm
    if (has_imm_.load(std::memory_order_relaxed)) {
      const uint64_t imm_start = env_->NowMicros();
      mutex_.Lock();
      if (imm_ != nullptr) {
        CompactMemTable(); // 这个函数可以压缩不可变内存表
        // Wake up MakeRoomForWrite() if necessary.若imm为空格，唤醒相关函数
        background_work_finished_signal_.SignalAll();
      }
      mutex_.Unlock();
      imm_micros += (env_->NowMicros() - imm_start); // 记录处理不可变内存表所花费的时间
    }

    Slice key = input->key();// 通过迭代器获取当前键
    if (compact->compaction->ShouldStopBefore(key) &&
        compact->builder != nullptr) { // 到达停止条件时完成写入
      status = FinishCompactionOutputFile(compact, input);
      if (!status.ok()) {
        break;
      }
    }

    // Handle key/value, add to state, etc.
    bool drop = false; // 解析当前键，获取用户键和序列号
    if (!ParseInternalKey(key, &ikey)) {
      // Do not hide error keys 解析失败，重置相关变量
      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) {
        // First occurrence of this user 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) {
        // Hidden by an newer entry for same user key
        drop = true;  // (A)
      } else if (ikey.type == kTypeDeletion &&
                 ikey.sequence <= compact->smallest_snapshot &&
                 compact->compaction->IsBaseLevelForKey(ikey.user_key)) {
        // For this user key:
        // (1) there is no data in higher levels
        // (2) data in lower levels will have larger sequence numbers
        // (3) data in layers that are being compacted here and have
        //     smaller sequence numbers will be dropped in the next
        //     few iterations of this loop (by rule (A) above).
        // Therefore this deletion marker is obsolete and can be dropped.
        drop = true;
      }

      last_sequence_for_key = ikey.sequence;
    }

    if (!drop) { // 继续处理
      /*打开输出文件，更新输出文件最小最大键，将键值对添加到当前输出文件*/
      // Open output file if necessary
      if (compact->builder == nullptr) {
        status = OpenCompactionOutputFile(compact);
        if (!status.ok()) {
          break;
        }
      }
      // 如果为第一个条目，设置最小值
      if (compact->builder->NumEntries() == 0) {
        compact->current_output()->smallest.DecodeFrom(key);
      } // 更新当前条目的最大键
      compact->current_output()->largest.DecodeFrom(key);
      compact->builder->Add(key, input->value()); // 添加键值对

      // Close output file if it is big enough 关闭输出文件
      if (compact->builder->FileSize() >=
          compact->compaction->MaxOutputFileSize()) {
        status = FinishCompactionOutputFile(compact, input);
        if (!status.ok()) {
          break;
        }
      }
    }

    input->Next();
  }

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
  /*处理完成情况*/
  if (status.ok() && shutting_down_.load(std::memory_order_acquire)) {
    status = Status::IOError("Deleting DB during compaction");
  }
  if (status.ok() && compact->builder != nullptr) {
    status = FinishCompactionOutputFile(compact, input);
  }
  if (status.ok()) {
    status = input->status();
  }
  delete input;
  input = nullptr;
  /*统计信息：包括压缩时间，字节读取，写入统计*/
  CompactionStats stats;
  stats.micros = env_->NowMicros() - start_micros - imm_micros;
  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;
  }

  mutex_.Lock();
  stats_[compact->compaction->level() + 1].Add(stats);
  /*安装压缩结果：更新版本控制，记录错误*/
  if (status.ok()) {
    status = InstallCompactionResults(compact);
  }
  if (!status.ok()) {
    RecordBackgroundError(status);
  }
  /*记录日志*/
  VersionSet::LevelSummaryStorage tmp;
  Log(options_.info_log, "compacted to: %s", versions_->LevelSummary(&tmp));
  return status;
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
class Compaction {
 public:
  ...
 private:
  friend class Version;
  friend class VersionSet;

  Compaction(const Options* options, int level);

  int level_; //正在进行压缩的级别
  uint64_t max_output_file_size_; // 最大输出文件大小限制
  Version* input_version_; // 指向当前压缩的输入版本
  VersionEdit edit_; // 保存压缩操作所作的编辑内容，包含需要添加和删除的文件

  // Each compaction reads inputs from "level_" and "level_+1"
  // 一个存储当前级别，一个存储下一个几倍
  std::vector<FileMetaData*> inputs_[2];  // The two sets of inputs

  // State used to check for number of overlapping grandparent files
  // (parent == level_ + 1, grandparent == level_ + 2)
  std::vector<FileMetaData*> grandparents_;
  size_t grandparent_index_;  // Index in grandparent_starts_
  bool seen_key_;             // Some output key has been seen
  int64_t overlapped_bytes_;  // Bytes of overlap between current output
                              // and grandparent files

  // State for implementing IsBaseLevelForKey

  // level_ptrs_ holds indices into input_version_->levels_: our state
  // is that we are positioned at one of the file ranges for each
  // higher level than the ones involved in this compaction (i.e. for
  // all L >= level_ + 2).
  // 输入版本的文件列表中为每个级别保存指向当前状态的索引
  size_t level_ptrs_[config::kNumLevels];
};

1
2
3
4
5
6
7
8
9
10
11
enum Tag {
  kComparator = 1, // 比较器的名称字符串
  kLogNumber = 2, // 文件编号
  kNextFileNumber = 3, // 下一个文件分配的编号
  kLastSequence = 4, // 当前版本的最后一个序列号
  kCompactPointer = 5, // 压缩指针
  kDeletedFile = 6, // 删除的元数据
  kNewFile = 7, // 新文件
  // 8 was used for large value refs
  kPrevLogNumber = 9 // 文件编号
};

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
/*文件元数据，存储每个sst文件的元数据*/
struct FileMetaData {
  FileMetaData() : refs(0), allowed_seeks(1 << 30), file_size(0) {}

  int refs; // 引用计数
  int allowed_seeks;  // Seeks allowed until compaction 容易触发seek compaction
  uint64_t number; // 文件的唯一编号
  uint64_t file_size;    // File size in bytes
  InternalKey smallest;  // Smallest internal key served by table 
  InternalKey largest;   // Largest internal key served by table
};

/*记录版本的变化信息*/
class VersionEdit {
 public:
  VersionEdit() { Clear(); }
  ~VersionEdit() = default;
  void Clear();
  // 设置信息
  void SetComparatorName(const Slice& name) {
    has_comparator_ = true;
    comparator_ = name.ToString();
  }
  void SetLogNumber(uint64_t num) {
    has_log_number_ = true;
    log_number_ = num;
  }
  void SetPrevLogNumber(uint64_t num) {
    has_prev_log_number_ = true;
    prev_log_number_ = num;
  }
  void SetNextFile(uint64_t num) {
    has_next_file_number_ = true;
    next_file_number_ = num;
  }
  void SetLastSequence(SequenceNumber seq) {
    has_last_sequence_ = true;
    last_sequence_ = seq;
  }
  void SetCompactPointer(int level, const InternalKey& key) {
    compact_pointers_.push_back(std::make_pair(level, key));
  }
  // 添加删除文件
  // Add the specified file at the specified number.
  // REQUIRES: This version has not been saved (see VersionSet::SaveTo)
  // REQUIRES: "smallest" and "largest" are smallest and largest keys in file
  void AddFile(int level, uint64_t file, uint64_t file_size,
               const InternalKey& smallest, const InternalKey& largest) {
    FileMetaData f;
    f.number = file;
    f.file_size = file_size;
    f.smallest = smallest;
    f.largest = largest;
    new_files_.push_back(std::make_pair(level, f));
  }
  // Delete the specified "file" from the specified "level".
  void RemoveFile(int level, uint64_t file) {
    deleted_files_.insert(std::make_pair(level, file));
  }
  // 编码解码
  void EncodeTo(std::string* dst) const;
  Status DecodeFrom(const Slice& src);

  std::string DebugString() const;

 private:
  friend class VersionSet;

  typedef std::set<std::pair<int, uint64_t>> DeletedFileSet;

  std::string comparator_; //比较器名称
  uint64_t log_number_; // 日志编号
  uint64_t prev_log_number_;
  uint64_t next_file_number_;
  SequenceNumber last_sequence_; // 最后一个序列号
  bool has_comparator_;
  bool has_log_number_;
  bool has_prev_log_number_;
  bool has_next_file_number_;
  bool has_last_sequence_;

  std::vector<std::pair<int, InternalKey>> compact_pointers_; // 存储压缩指针
  DeletedFileSet deleted_files_; // 跟踪已删除的文件，便于版本更新时管理
  std::vector<std::pair<int, FileMetaData>> new_files_; // 存储新添加的文件信息
};

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
class Version{
public:
    ...
 private:
  friend class Compaction;
  friend class VersionSet;

  class LevelFileNumIterator;

  explicit Version(VersionSet* vset)
      : vset_(vset),
        next_(this),
        prev_(this),
        refs_(0),
        file_to_compact_(nullptr),
        file_to_compact_level_(-1),
        compaction_score_(-1),
        compaction_level_(-1) {}

  Version(const Version&) = delete;
  Version& operator=(const Version&) = delete;

  ~Version();

  Iterator* NewConcatenatingIterator(const ReadOptions&, int level) const;

  // Call func(arg, level, f) for every file that overlaps user_key in
  // order from newest to oldest.  If an invocation of func returns
  // false, makes no more calls.
  //
  // REQUIRES: user portion of internal_key == user_key.
  void ForEachOverlapping(Slice user_key, Slice internal_key, void* arg, bool (*func)(void*, int, FileMetaData*));

  VersionSet* vset_;  // VersionSet to which this Version belongs
  Version* next_;     // Next version in linked list
  Version* prev_;     // Previous version in linked list
  int refs_;          // Number of live refs to this version

  // List of files per level
  std::vector<FileMetaData*> files_[config::kNumLevels];

  // Next file to compact based on seek stats.
  FileMetaData* file_to_compact_;
  int file_to_compact_level_;

  // Level that should be compacted next and its compaction score.
  // Score < 1 means compaction is not strictly needed.  These fields
  // are initialized by Finalize().
  double compaction_score_;
  int compaction_level_;
};

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
class VersionSet {
public:
    ...
 private:
  class Builder;

  friend class Compaction;
  friend class Version;

  bool ReuseManifest(const std::string& dscname, const std::string& dscbase);

  void Finalize(Version* v);
    // 进行压缩或者查询时，了解数据范围有助于优化性能
  void GetRange(const std::vector<FileMetaData*>& inputs, InternalKey* smallest, InternalKey* largest);
    // 合并文件时考虑文件范围
  void GetRange2(const std::vector<FileMetaData*>& inputs1, const std::vector<FileMetaData*>& inputs2, InternalKey* smallest, InternalKey* largest);
    // 为压缩操作设置其它输入文件
  void SetupOtherInputs(Compaction* c);

  // Save current contents to *log
  Status WriteSnapshot(log::Writer* log);

  void AppendVersion(Version* v);

  Env* const env_; 
  const std::string dbname_; // 数据库名称
  const Options* const options_; // 数据库选项
  TableCache* const table_cache_; // 表缓存
  const InternalKeyComparator icmp_; // 内部键比较器
  uint64_t next_file_number_;
  uint64_t manifest_file_number_;
  uint64_t last_sequence_; // 所有写操作中最后使用的序列号
  uint64_t log_number_;
  uint64_t prev_log_number_;  // 0 or backing store for memtable being compacted

  // Opened lazily
  WritableFile* descriptor_file_;  // 数据库描述符文件
  log::Writer* descriptor_log_; // 写入日志
  Version dummy_versions_;  // Head of circular doubly-linked list of versions.循环双链表的虚拟头
  Version* current_;        // == dummy_versions_.prev_指向当前版本

  // Per-level key at which the next compaction at that level should start.压缩指针
  // Either an empty string, or a valid InternalKey.
  std::string compact_pointer_[config::kNumLevels];
};

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
  Status Run() {
    /*查找数据库目录中的文件
    * 解析文件名，提取文件编号和类型
    * 对于描述符文件，文件名存入manifest
    * 对于日志和表文件，存储在logs和table_number中
    */
    Status status = FindFiles(); 
    if (status.ok()) {
      ConvertLogFilesToTables(); // 将日志文件转换成表文件
      ExtractMetaData(); // 获取每个文件的元数据(key)以及最终数据库的元数据(seq)
      status = WriteDescriptor(); // 创建一个新的manifest文件，将扫描的数据库元数据进行记录
    }
    if (status.ok()) {
      unsigned long long bytes = 0;
      for (size_t i = 0; i < tables_.size(); i++) {
        bytes += tables_[i].meta.file_size;
      }
      Log(options_.info_log,
          "**** Repaired leveldb %s; "
          "recovered %d files; %llu bytes. "
          "Some data may have been lost. "
          "****",
          dbname_.c_str(), static_cast<int>(tables_.size()), bytes);
    }
    return status;
  }