[LevelDB]关于LevelDB存储架构到底怎么设计的？

LevelDB 存储架构重要特点

1. LSM树结构：本质上就是分层存储+WAL+后台进程异步写入，将原本需要任意写入(random write: 我觉得翻译成任意写入更合理，随机写入是传统叫法)
   如下图架构

// LSM整体架构
+-------------------------+
|       Write Path        |
+-------------------------+
        ↓
+-------------------------+
|    Write Ahead Log      |  // WAL日志
+-------------------------+
        ↓
+-------------------------+
|      MemTable (L0)      |  // 活跃内存表（跳表）
+-------------------------+
        ↓
+-------------------------+
|  Immutable MemTable     |  // 不可变内存表
+-------------------------+
        ↓
+-------------------------+
|   SSTable Files (L0)    |  // Level 0：文件可能重叠
+-------------------------+
        ↓
+-------------------------+
|   SSTable Files (L1)    |  // Level 1：文件有序不重叠
+-------------------------+
        ↓
+-------------------------+
|   SSTable Files (L2)    |  // Level 2：容量是L1的10倍
+-------------------------+
        ↓
+-------------------------+
|   SSTable Files (L3)    |  // Level 3：容量是L2的10倍
+-------------------------+

WAL日志：指的是操作日志，提前写入操作日志，是为了保证操作的原子化，如果在写入过程中出现问题,能够根据WAL日志重新运行操作
MemTable：使用跳表数据结构来实现内存结构
SSTable Files: 在磁盘上的文件，如果在实际的运行中，本质上指的是运行LevelDB的磁盘日志
todo: 缺个图

2. 压缩策略： 使用可选择的压缩算法，对数据流进行可控改造，本质上是平衡计算机CPU和IO，因为压缩是相当于用一种编码信息的方式对原有数据进行精简表达。
3. 索引加速: 使用布隆过滤对数据进行快速索引，就是相当于用多个hash来算同一个值，从而避免
4. 自主控制的内存: C++相比Java来说最重要的变化就是，Java所有的代码都是run在JVM上的，就是相当于会把Java代码再转化成.class文件，然后再被JVM进行解释，这样JVM就会代替app层来进行内存管理，但是坏处是本质上会多消耗一些资源。C++则主要通过自己来进行内存的管理和控制，本质上就是通过创建对象并通过delete来进行控制对象的创建和销毁，虽然现在出现智能指针能够自己来控制，但是自己来控制的代价本质上和JVM一样会增大内存的消耗，但是如果忘记销毁会出现潜在内存泄漏，程序会run着run着就挂了，隔一段时间要重启一次(别问我怎么知道的)
   PS: LevelDB实际实现的时候有挺多特点，这里只挑选最重要几个

总体概括

本文准备从自上而下来解释LevelDB的存储设计结构（后续可能会补充和其他存储结构的异同todo，现在还不会）

首先我们要明白上图中主要提出了这样几个概念和实体，分别一句话解释

1. USER:用户
2. MemTable：能够被写入的可变内存
3. Immutable：MemTable如果写满就不可写入，后续异步写入磁盘
4. table: 本质上就是在磁盘上的一个sstable文件，就是运行了leveldb程序后,写入数据后生成的日志文件
5. block: LevelDB的最小存储单元,多个block组成了table
6. block Cache/Table Cache：对表(table)和块(block)进行缓存

LevelDB中内存模型

首先从内存的视角来看
主要分为下面这些模型

1. Arena：高效的小对象分配
2. Block Cache：读取性能优化
3. TableCache：文件句柄管理
4. MemTable：写入性能优化

PS：MemTable&Immutable 这两个实体的本质上是一样的，只是Immutable是Memtable的不可变的版本，接下来准备从下面两个方面来解释这两个实体

MemTable

MemTable的数据结构

背景：

MemTable使用Skiplist(跳表)数据结构来加速在内存结构对key的检索，本质上是通过Skiplist这种跳表数据结构来优化kv索引中k的遍历。

SkipList

Skiplist的数据结构

本质上SkipList 由 Node元素组成，具体的组成方式如下图所示，HEAD蓝块和黄色块都是Node, 这里的L3-L1指的是 Skiplist中的层级，因为SkipList是一个跳表数据结构，会根据不同的key来跳过一些node，来加速索引，

1. key：当前
   

Skiplist的数据访问细节

具体查找过程如下：

1. 从头节点（HEAD）的最高层（L3）开始查找
2. 在L3层向右移动到key=30的节点（发现下一个节点会超过目标值）
3. 从key=30的L3层下降到L2层继续查找
4. 从key=30节点降至L2层
5. 从L3层降至L2层继续查找
6. 从L3层降到L2层 （30 < 40，但30的右节点值会 > 40）
7. 在L2层向右找到目标节点key=40”

参考源代码

SkipList<Key, Comparator>::FindGreaterOrEqual(const Key& key,
                                              Node** prev) const {
  // 从头节点开始
  Node* x = head_;
  // 从最高层开始
  int level = GetMaxHeight() - 1;
  while (true) {
    // 获取当前层的下一个节点
    Node* next = x->Next(level);
    // 如果当前层的下一个节点大于key, 则继续向下层查找
    if (KeyIsAfterNode(key, next)) {
      // Keep searching in this list
      x = next;
    } else {
      // 如果prev不为空, 则将当前节点赋值给prev[level]
      if (prev != nullptr) prev[level] = x;
      // 如果当前层为0层, 则返回下一个节点
      if (level == 0) {
        return next;
      } else {
        // 如果当前层不是0层, 则继续向下层查找  
        level--;
      }
    }
  }
}

SkipList的核心方法

在添加新节点时，会通过如下的方法确定当前node节点的高度，即height
概率分布：

1. 所有节点(100%)至少有高度1
2. 约25%的节点高度≥2
3. 约6.25%的节点高度≥3
4. 约1.56%的节点高度≥4
5. 以此类推…

template <typename Key, class Comparator>
int SkipList<Key, Comparator>::RandomHeight() {
  static const unsigned int kBranching = 4;// kBranching = 4 表示概率因子
  int height = 1;
  while (height < kMaxHeight && rnd_.OneIn(kBranching)) {// 相当于有1/4 概率为true
    height++;
  }
  assert(height > 0);
  assert(height <= kMaxHeight);
  return height;
}

Node细节

Node主要有两个关键变量

1. key: KV数据库里面的key
2. next_： 表示当前的节点(Node)的下一个节点的地址

Next和NoBarrier_Next 这两个方法的区别就是，一个的next_变量的访问是需要等到所有的内存写操作结束之后才会进行读取，NoBarrier_Next 直接会读取原子变量 next_ 不会有任何等待

源代码

// 接下来 是具体的实现细节:  嵌套结构体 Node 的实现
template <typename Key, class Comparator>
struct SkipList<Key, Comparator>::Node {

  explicit Node(const Key& k) : key(k) {}

  Key const key;

  // 链接的访问器/修改器。封装在方法中，以便我们可以
  // 根据需要添加适当的内存屏障。
  Node* Next(int n) {
    assert(n >= 0);
    // 使用“获取加载”操作，以便我们观察返回的Node的完全初始化版本。
    // 表示所有写操作都在更新指针前执行完
    return next_[n].load(std::memory_order_acquire);
  }
  void SetNext(int n, Node* x) {
    assert(n >= 0);
    // 使用“释放存储”，以便任何通过此指针读取的线程都会看到一个完全初始化的版本。
    // 表示所有写操作都在更新指针后执行完
    next_[n].store(x, std::memory_order_release);
  }

  // 无内存屏障的变体，可以在少数位置安全使用。
  Node* NoBarrier_Next(int n) {
    assert(n >= 0);
    return next_[n].load(std::memory_order_relaxed);// 使用std::memory_order_relaxed 不会
  }
  void NoBarrier_SetNext(int n, Node* x) {
    assert(n >= 0);
    next_[n].store(x, std::memory_order_relaxed);
  }

 private:
  // Array of length equal to the node height.  next_[0] is lowest level link.
  std::atomic<Node*> next_[1];
};

MemTable的数据加速方式

MemTable的数据访问加速通过Iterator这个特性来进行实现， 为什么需要Iterator，而不是通过Node._next直接实现数据访问，主要是使用Iterator可以相比直接使用Node的方法有以下的额外特性，本质上就是对SKiplist的数据结构扩展一些能够方便遍历的特性。

1. 双向遍历
2. 状态管理， Valid()方法

Iterator 的核心方法

1. Next： 向后遍历,本质上使用node_.Next(0) 。
2. Prev： 向前遍历, 本质上相当于 list_->FindLessThan(node_->key)。

template <typename Key, class Comparator>
// 判断当前的迭代器是否有效，如果有效就调用next方法
inline void SkipList<Key, Comparator>::Iterator::Next() {
  assert(Valid());
  node_ = node_->Next(0);
}

template <typename Key, class Comparator>
// 判断当前的迭代器是否有效，如果有效就调用prev方法
inline void SkipList<Key, Comparator>::Iterator::Prev() {
  // Instead of using explicit "prev" links, we just search for the
  // last node that falls before key.
  assert(Valid());
  // 调用 findLessThan 方法，找到当前节点的前一个节点
  node_ = list_->FindLessThan(node_->key);
  if (node_ == list_->head_) {
    node_ = nullptr;
  }
}

MemTable 的读取&写入细节

主要的方法有以下

1. Add: 写入MemTable的方法
   

2. 计算大小

3. 计算总编码长度

4. 内存分配

5. 编码与复制数据

6. 验证与插入

7. Get: 读取MemTable的方法
   
   流程

8. 迭代器是否找到有效位置

9. 键前缀是否匹配

10. 根据类型标记确定是返回值还是返回已删除状态

void MemTable::Add(SequenceNumber s, ValueType type, const Slice& key,
                   const Slice& value) {
  // 这里可以看到 具体内存中数据的实现, 这里相当于做了 key 和 value的一层转换
  /**
   * 这里的重点就是 arena_.Allocate 分配内存
   * std::memcpy: 内存复制
   */
  // Format of an entry is concatenation of:
  //  key_size     : varint32 of internal_key.size()
  //  key bytes    : char[internal_key.size()]
  //  tag          : uint64((sequence << 8) | type)
  //  value_size   : varint32 of value.size()
  //  value bytes  : char[value.size()]
  size_t key_size = key.size();
  size_t val_size = value.size();
  size_t internal_key_size = key_size + 8;
  const size_t encoded_len = VarintLength(internal_key_size) +
                             internal_key_size + VarintLength(val_size) +
                             val_size;
  char* buf = arena_.Allocate(encoded_len);// 使用arena 来进行内存分配
  char* p = EncodeVarint32(buf, internal_key_size);
  std::memcpy(p, key.data(), key_size);
  p += key_size;
  EncodeFixed64(p, (s << 8) | type);// 这个是为了内存对齐, todo, 确认下
  p += 8;
  p = EncodeVarint32(p, val_size);
  std::memcpy(p, value.data(), val_size);
  assert(p + val_size == buf + encoded_len);
  table_.Insert(buf); // 这里还是使用跳表来进行存储, 使用跳表来组织结构
}


bool MemTable::Get(const LookupKey& key, std::string* value, Status* s) {// Memtable 的get 方法, 相当于内存进行了一层缓存
  Slice memkey = key.memtable_key();
  Table::Iterator iter(&table_);
  iter.Seek(memkey.data());
  if (iter.Valid()) {
    // entry format is:
    //    klength  varint32
    //    userkey  char[klength]
    //    tag      uint64
    //    vlength  varint32
    //    value    char[vlength]
    const char* entry = iter.key();
    uint32_t key_length;
    const char* key_ptr = GetVarint32Ptr(entry, entry + 5, &key_length);
    if (comparator_.comparator.user_comparator()->Compare(
            Slice(key_ptr, key_length - 8), key.user_key()) == 0) {
      // Correct user key
      const uint64_t tag = DecodeFixed64(key_ptr + key_length - 8);
      switch (static_cast<ValueType>(tag & 0xff)) {
        case kTypeValue: {
          Slice v = GetLengthPrefixedSlice(key_ptr + key_length);
          value->assign(v.data(), v.size());
          return true;
        }
        case kTypeDeletion:
          *s = Status::NotFound(Slice());
          return true;
      }
    }
  }
  return false

MemTable的内存分配器 --Arena

作用: 为memtable这种对象，比如说频繁插入内存的情况下，提前申请一批内存来防止直接使用 malloc 和new调用系统调用，这样的话开销非常高,并且标准实现中内存会存在全局锁，会有内存竞争的问题，这里需要对比一下 ，本质上是使用 池化的计数来对内存进行管理

Arena 核心方法

1. Allocate: 从当前内存块中快速分配内存，空间不足时调用AllocateFallback。
2. AllocateFallback: 处理内存不足情况，根据请求大小决定分配专用块或新标准块。
3. AllocateAligned: 分配内存时确保地址对齐，计算额外填充并处理对齐要求。

Arena 核心方法源代码

// 使用内联防止内存展开, 主要的作用是降低内存开销
inline char* Arena::Allocate(size_t bytes) {
  // The semantics of what to return are a bit messy if we allow
  // 0-byte allocations, so we disallow them here (we don't need
  // them for our internal use).
  assert(bytes > 0);
  if (bytes <= alloc_bytes_remaining_) {
    char* result = alloc_ptr_;
    alloc_ptr_ += bytes;
    alloc_bytes_remaining_ -= bytes;
    return result;
  }
  return AllocateFallback(bytes);
}

// 当 当前块内存不足的时候
char* Arena::AllocateFallback(size_t bytes) {
  if (bytes > kBlockSize / 4) {
    // Object is more than a quarter of our block size.  Allocate it separately
    // to avoid wasting too much space in leftover bytes.
    char* result = AllocateNewBlock(bytes);
    return result;
  }

  // We waste the remaining space in the current block.
  alloc_ptr_ = AllocateNewBlock(kBlockSize);
  alloc_bytes_remaining_ = kBlockSize;

  char* result = alloc_ptr_;
  alloc_ptr_ += bytes;
  alloc_bytes_remaining_ -= bytes;
  return result;
}

char* Arena::AllocateAligned(size_t bytes) {
  const int align = (sizeof(void*) > 8) ? sizeof(void*) : 8;// 判断一个指针需要多少字节? 从而在不同的机器上进行内存对齐
  static_assert((align & (align - 1)) == 0,
                "Pointer size should be a power of 2");
  size_t current_mod = reinterpret_cast<uintptr_t>(alloc_ptr_) & (align - 1);
  size_t slop = (current_mod == 0 ? 0 : align - current_mod);
  size_t needed = bytes + slop;
  char* result;
  if (needed <= alloc_bytes_remaining_) {
    result = alloc_ptr_ + slop;
    alloc_ptr_ += needed;
    alloc_bytes_remaining_ -= needed;
  } else {
    // AllocateFallback always returned aligned memory
    result = AllocateFallback(bytes);
  }
  assert((reinterpret_cast<uintptr_t>(result) & (align - 1)) == 0);
  return result;
}

Table

Table的本质就是对应磁盘文件中的sstable，是由多个 Block组成，组成的Block主要有以下部分

1. 数据块(Data blocks)- 存储实际的键值对
2. 索引
3. 元数据块
4. 过滤器块
5. 页脚

Table的交互逻辑主要有两个部分

1. TableBuilder：写入逻辑
2. Table：读取逻辑

Table 对外提供服务场景

Table 抽象主要有三种场景会调用当前的Table，写入压缩读取

写入场景

• 用户写请求 → DBImpl::Write → MemTable填充 → CompactMemTable → BuildTable → TableBuilder

/**
 * 压缩内存表,将不可变内存表压缩为新的表文件，并更新版本集。
 */
void DBImpl::CompactMemTable() {
 ....
 //***************************************************************************************************
  Status s = WriteLevel0Table(imm_, &edit, base);// 写入一个新的 Level0表文件, 更新edit 将内存表转换成SSTable
// ***************************************************************************************************
 ...
} 
/*
 * 这个方法用来写入Level0等级的表
 */
Status DBImpl::WriteLevel0Table(MemTable* mem, VersionEdit* edit,
                                Version* base) {
 
// ...
// ************************************************************************************************
    s = BuildTable(dbname_, env_, options_, table_cache_, iter, &meta);// 通过BuildTable方法构建 Table对象
// ************************************************************************************************
// ...
}

如上代码 先调用CompactMemTable方法，然后调用WriteLevel0Table方法，接着调用BuildTable方法

压缩场景

• 后台Compaction → DoCompactionWork → BuildTable → TableBuilder

读取场景

读取场景主要被用在TableCache中的Table::InternalGet方法

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;
 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); // Table对外暴露方法
    // **************************************************************************************************
   cache_->Release(handle);
 }
 return s;
}

• 用户读请求 → DBImpl::Get → Version::Get → TableCache → Table::InternalGet

Table写入逻辑-TableBuilder

写入细节

1. 输入处理：方法接收一对 key-value 作为输入
2. 索引块处理：

• 检查是否有待处理的索引条目（pending_index_entry）
• 如果有，使用比较器找到上一个 key 和当前 key 的最短分隔符
• 将分隔符和对应的 handle 编码后添加到索引块中

3. 过滤块处理：

• 如果过滤块存在，则调用 filter_block->AddKey(key) 将 key 添加到过滤块

4. 数据块处理：

• 更新 last_key 和条目计数 num_entries
• 将 key-value 对添加到数据块
• 估计当前数据块大小，如果超过设定阈值，调用 Flush() 方法

5. Flush() 会将当前数据块写入 SSTable 文件

源代码

void TableBuilder::Add(const Slice& key, const Slice& value) {
  Rep* r = rep_;
  // rep 是否close了
  assert(!r->closed);
  if (!ok()) return;
  if (r->num_entries > 0) {
    assert(r->options.comparator->Compare(key, Slice(r->last_key)) > 0);
  }
  // 处理 待处理的索引条目
  if (r->pending_index_entry) {
    assert(r->data_block.empty());// 数据块 是空的
    r->options.comparator->FindShortestSeparator(&r->last_key, key);// 找到 最后一个key 和当前key的最短分隔符
    std::string handle_encoding;
    r->pending_handle.EncodeTo(&handle_encoding);
    r->index_block.Add(r->last_key, Slice(handle_encoding));// 索引块添加 
    r->pending_index_entry = false;
  }
  // 
  if (r->filter_block != nullptr) {
    r->filter_block->AddKey(key);// 添加过滤块
  }
  // 将string类型的数据 重新设置
  r->last_key.assign(key.data(), key.size());
  r->num_entries++;// 
  r->data_block.Add(key, value);//  数据块添加

  const size_t estimated_block_size = r->data_block.CurrentSizeEstimate();
  if (estimated_block_size >= r->options.block_size) {//判断块大小
    Flush();// 调用Flush方法 把数据写入到内存中
  }
}

Table读取逻辑-Table

Table.InternalGet 方法，调用Get方法来读取Table的内容

查询细节

1. 输入的 Key 首先进入索引块，通过 iiter->Seek(k) 定位可能包含目标键的数据块
2. 通过布隆过滤器的 KeyMayMatch 快速判断键是否可能存在
3. 最后在数据块中通过 block_iter->Seek(k) 精确查找键值对
4. 如果找到目标键值对，调用 handle_result(arg, key, value) 处理结果

源代码

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()) {
    // 找到可能包含目标key的位置
    Slice handle_value = iiter->value();
    // 布隆过滤器检查
    FilterBlockReader* filter = rep_->filter;
    BlockHandle handle;
    if (filter != nullptr && handle.DecodeFrom(&handle_value).ok() &&
        !filter->KeyMayMatch(handle.offset(),
                             k)) {  // 如果布隆过滤器表示key不存在，直接返回
      // Not found
    } else {
      // 读取实际的数据块
      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;
}

Block

主要就是为了支撑 Table的检索，本质上 Table是对外提供了一个sstable的可用抽象，而block是对sstable做更细粒度的管理。

Block从写入写出流程来看，分为两个部分

1. Block: 读取流程
2. BlockBuilder: 写入流程

Block 的读取流程

Block 数据结构和查找流程:
┌─────────────────────────────────── Block ────────────────────────────────────┐
│                                                                              │
│  ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐                 │
│  │ Restart │ │ Restart │ │ Restart │ │ Restart │ │ Restart │   数据区域       │
│  │ Point 0 │ │ Point 1 │ │ Point 2 │ │ Point 3 │ │ Point 4 │             │
│  └─────────┘ └─────────┘ └─────────┘ └─────────┘ └─────────┘             │
│       │           │           │           │           │                     │
│       ▼           ▼           ▼           ▼           ▼                     │
│   ┌─────────┬─────────┬─────────┬─────────┬─────────┐                     │
│   │  数据1  │  数据2  │  数据3  │  数据4  │  数据5  │                     │
│   └─────────┴─────────┴─────────┴─────────┴─────────┘                     │
│                                                                             │
└─────────────────────────────────────────────────────────────────────────────┘

查找过程示意 (假设查找 "apple"):

1. 初始化搜索范围:
left = 0                right = 4
   │                       │
   ▼                       ▼
┌─────┬─────┬─────┬─────┬─────┐
│  0  │  1  │  2  │  3  │  4  │
└─────┴─────┴─────┴─────┴─────┘

2. 二分查找过程:
第一次迭代:
mid = (0 + 4 + 1) / 2 = 2
   ┌────── mid ──────┐
   │                 ▼
┌─────┬─────┬─────┬─────┬─────┐
│  0  │  1  │  2  │  3  │  4  │
└─────┴─────┴─────┴─────┴─────┘

比较 mid_key 与 target:
如果 mid_key < target:  left = mid
如果 mid_key >= target: right = mid - 1

3. 定位到重启点后的线性查找:
┌─────────────── 重启点区域 ───────────────┐
│     ┌────► 线性查找                      │
│     │                                    │
├─────▼───┬──────────┬──────────┬─────────┤
│ entry 1 │ entry 2  │ entry 3  │ entry 4 │
└─────────┴──────────┴──────────┴─────────┘

每个 Entry 的格式:
┌──────────┬───────────┬────────────┬──────┬───────┐
│  shared  │non_shared │value_length│  key │ value │
└──────────┴───────────┴────────────┴──────┴───────┘

优化说明:
1. 当前位置优化:
┌─────────────────────┐
│ if (Valid()) {      │
│   使用当前位置作为  │ 
│   查找起点          │
└─────────────────────┘

2. 跳过重复查找优化:
┌─────────────────────┐
│ skip_seek 条件:     │
│ - 目标在当前区块    │
│ - 当前key较小       │
└─────────────────────┘

查找流程说明

• 首先检查当前位置，如果已经有效，可能用作查找起点
• 在重启点数组中进行二分查找，找到最后一个小于目标的重启点
• 从找到的重启点开始，进行线性查找直到找到第一个大于等于目标的键
• 使用优化机制避免不必要的查找操作
  关键优化点
• 利用当前位置作为可能的起点
• 二分查找快速定位重启区间
• 通过 skip_seek 优化避免重复查找
• 结合重启点和线性查找平衡查询效率

源代码

  void Seek(const Slice& target) override {
    // Binary search in restart array to find the last restart point
    // with a key < target
    uint32_t left = 0;
    uint32_t right = num_restarts_ - 1;
    int current_key_compare = 0;

    if (Valid()) {
      // If we're already scanning, use the current position as a starting
      // point. This is beneficial if the key we're seeking to is ahead of the
      // current position.
      current_key_compare = Compare(key_, target);
      if (current_key_compare < 0) {
        // key_ is smaller than target
        left = restart_index_;
      } else if (current_key_compare > 0) {
        right = restart_index_;
      } else {
        // We're seeking to the key we're already at.
        return;
      }
    }

    while (left < right) {
      uint32_t mid = (left + right + 1) / 2;
      uint32_t region_offset = GetRestartPoint(mid);
      uint32_t shared, non_shared, value_length;
      const char* key_ptr =
          DecodeEntry(data_ + region_offset, data_ + restarts_, &shared,
                      &non_shared, &value_length);
      if (key_ptr == nullptr || (shared != 0)) {
        CorruptionError();
        return;
      }
      Slice mid_key(key_ptr, non_shared);
      if (Compare(mid_key, target) < 0) {
        // Key at "mid" is smaller than "target".  Therefore all
        // blocks before "mid" are uninteresting.
        left = mid;
      } else {
        // Key at "mid" is >= "target".  Therefore all blocks at or
        // after "mid" are uninteresting.
        right = mid - 1;
      }
    }

    // We might be able to use our current position within the restart block.
    // This is true if we determined the key we desire is in the current block
    // and is after than the current key.
    assert(current_key_compare == 0 || Valid());
    bool skip_seek = left == restart_index_ && current_key_compare < 0;
    if (!skip_seek) {
      SeekToRestartPoint(left);
    }
    // Linear search (within restart block) for first key >= target
    while (true) {
      if (!ParseNextKey()) {
        return;
      }
      if (Compare(key_, target) >= 0) {
        return;
      }
    }
  }

Block 的写入流程

主要就是使用前缀压缩的方式减少存储的压力，具体的细节如下

键值对存储格式示意图:
┌──────────────────────────────────────────────────────────────────┐
│                        Block Buffer                              │
└──────────────────────────────────────────────────────────────────┘

示例: 添加两个键值对
1. 添加 "apple" -> "red"
┌─────────┬────────────┬────────────┬───────┬───────┐
│shared=0 │non_shared=5│value_size=3│"apple"│ "red" │
└─────────┴────────────┴────────────┴───────┴───────┘

2. 添加 "apply" -> "blue" (与"apple"共享"ap")
┌─────────┬────────────┬────────────┬───────┬────────┐
│shared=2 │non_shared=3│value_size=4│ "ply" │"blue"  │
└─────────┴────────────┴────────────┴───────┴────────┘

前缀压缩过程示意:
"apple" -> "red"
┌────────────────────┐
│ apple     -> red   │ (完整存储，因为是重启点)
└────────────────────┘

"apply" -> "blue"
┌────────────────────┐
│ ap[共享] + ply     │ (存储时只需要存储非共享部分"ply")
└────────────────────┘

数据结构状态:
counter_ : 计数器，到达 block_restart_interval 时重置
┌───────┐
│   1   │ (添加"apple"后)
└───────┘
┌───────┐
│   2   │ (添加"apply"后)
└───────┘

重启点数组 (restarts_):
┌───────┐
│   0   │ (第一个键的位置)
└───────┘

buffer_ 实际存储格式:
┌──────┬───────┬──────┬───────┬────┬──────┬───────┬──────┬────┬────────┐
│shared│non_sh.│val_sz│ key   │value│shared│non_sh.│val_sz│key │ value  │
│  0   │   5   │  3   │"apple"│"red"│  2   │   3   │  4   │"ply"│"blue" │
└──────┴───────┴──────┴───────┴────┴──────┴───────┴──────┴────┴────────┘

源代码

void BlockBuilder::Add(const Slice& key, const Slice& value) {
  Slice last_key_piece(last_key_);
  // 
  assert(!finished_);
  assert(counter_ <= options_->block_restart_interval);
  assert(buffer_.empty()  // No values yet?
         || options_->comparator->Compare(key, last_key_piece) > 0);
  // 
  size_t shared = 0;
  if (counter_ < options_->block_restart_interval) {
    const size_t min_length = std::min(last_key_piece.size(), key.size());
    while ((shared < min_length) && (last_key_piece[shared] == key[shared])) {
      shared++;
    }
  } else {
    // Restart compression
    restarts_.push_back(buffer_.size());
    counter_ = 0;
  }
  // 
  const size_t non_shared = key.size() - shared;

  // Add "<shared><non_shared><value_size>" to buffer_
  PutVarint32(&buffer_, shared);
  PutVarint32(&buffer_, non_shared);
  PutVarint32(&buffer_, value.size());

  // Add string delta to buffer_ followed by value
  buffer_.append(key.data() + shared, non_shared);
  buffer_.append(value.data(), value.size());

  // Update state
  last_key_.resize(shared);
  last_key_.append(key.data() + shared, non_shared);
  assert(Slice(last_key_) == key);
  counter_++;
}

猜你喜欢

C++多线程： https://blog.csdn.net/luog_aiyu/article/details/145548529
一文了解LevelDB数据库读取流程：https://blog.csdn.net/luog_aiyu/article/details/145946636
一文了解LevelDB数据库写入流程：https://blog.csdn.net/luog_aiyu/article/details/145917173

PS

你的赞是我很大的鼓励
欢迎大家加我飞书扩列, 希望能认识一些新朋友~
二维码见： https://www.cnblogs.com/DarkChink/p/18598402