作为LevelDB源码分析系列的第一篇文章，介绍Memtable的实现，以及其中涉及到的数据结构和辅助函数。

Memtable是LevelDB的内存数据结构。当一个Memtable满之后，会被变成Immutable Memtable，并写入SSTable Level0。

基本概念

VarInt

LevelDB 的 VarInt 机制，用一个 char 数组存放整数，主要目的不是支持大整数，而是压缩小整数的存储空间。例如小于128的 unsigned int，只要一个字节就行，但如果数比较大，。
具体实现就是读 char 数组，如果最高位是1，那么就说明这个 VarInt 还没有结束，于是就接着读下一位。
GetVarint32Ptr 的函数签名，传入的 limit 表示这个 VarInt 数组有多长。因为 VarInt 最多占用5个字节，所以一般传入都是 p + 5。需要注意，合法的 limit 应当始终大于 p。
一开始是一个优化分支，如果这个数组长度是1，那么最高位就是0，所以可以直接返回。否则调用 GetVarint32PtrFallback，那就是很直白的边移位边加上去。
这个函数会把要读的结果写到传入的 value 中，返回的前进后的 p。

inline const char* GetVarint32Ptr(const char* p, const char* limit,
                                  uint32_t* value) {
  if (p < limit) {
    uint32_t result = *(reinterpret_cast<const uint8_t*>(p));
    if ((result & 128) == 0) {
      *value = result;
      return p + 1;
    }
  }
  return GetVarint32PtrFallback(p, limit, value);
}

const char* GetVarint32PtrFallback(const char* p, const char* limit,
                                   uint32_t* value) {
  uint32_t result = 0;
  for (uint32_t shift = 0; shift <= 28 && p < limit; shift += 7) {
    uint32_t byte = *(reinterpret_cast<const uint8_t*>(p));
    p++;
    if (byte & 128) {
      // More bytes are present
      result |= ((byte & 127) << shift);
    } else {
      result |= (byte << shift);
      *value = result;
      return reinterpret_cast<const char*>(p);
    }
  }
  return nullptr;
}

Slice

Slice 定义在 include/leveldb/slice.h 里面，可以理解为对const char*的View。

Memtable类

Memtable类定义在memtable.h/cc里面。接口如下

size_t ApproximateMemoryUsage()
估计大小，在被修改的时候也可以调用。
Iterator* NewIterator()
void Add(SequenceNumber seq, ValueType type, const Slice& key, const Slice& value)
负责插入记录。
1. type
  普通插入type为kTypeValue。
  删除操作实际上是插入type为kTypeDeletion的记录。
2. key
bool Get(const LookupKey& key, std::string* value, Status* s)
如果存在key对应的记录，返回true，并且将值存在*value上。
如果存在key被删除的记录，返回true，并且在*status存入NotFound()。
否则返回false。
这里LookupKey的设计较为复杂，稍后讲解。
void Ref()
void Unref()

Memtable还有下面一些成员：

KeyComparator comparator_
封装了个InternalKeyComparator，这一块后面介绍

struct KeyComparator {
  const InternalKeyComparator comparator;
  explicit KeyComparator(const InternalKeyComparator& c) : comparator(c) {}
  int operator()(const char* a, const char* b) const;
};

int refs_
Arena arena_

Table table_
实际上是个跳表。

1	typedef SkipList<const char*, KeyComparator> Table;

Key

Memtable 在跳表中索引的 Key 中存放三个数据：

User Key
用户实际传入的key。
Sequence Number
1
typedef uint64_t SequenceNumber;
Value Type
表示是不是删除。
对于C而言，enum的大小取决于里面定义的范围。对于C++，可以显式指定实际使用的类型。
1
enum ValueType { kTypeDeletion = 0x0, kTypeValue = 0x1 };

这些数据是按顺序编码的，这样方便比较。

InternalKey 把这三个数据打包到一个 std::string 里面。
ParseInternalKey 把 InternalKey 转成 ParsedInternalKey，后者可以直接读取上面三个字段。
AppendInternalKey 可以从 ParsedInternalKey 构建 InternalKey。

为了方便查找，还将 User Key 和 Sequence Number 合并，组成 LookupKey。

start_
指向 klength，klength 表示 userkey 长度。注意，这里 userkey 也可以存放 InternalKey，所以 LookupKey 可以表示一个 Memtable Key。

klength  varint32               <-- start_
userkey  char[klength]          <-- kstart_
tag      uint64
                                <-- end_

kstart_
end_

class LookupKey {
 public:
  // Initialize *this for looking up user_key at a snapshot with
  // the specified sequence number.
  LookupKey(const Slice& user_key, SequenceNumber sequence);

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

  ~LookupKey();

  // Return a key suitable for lookup in a MemTable.
  Slice memtable_key() const { return Slice(start_, end_ - start_); }

  // Return an internal key (suitable for passing to an internal iterator)
  Slice internal_key() const { return Slice(kstart_, end_ - kstart_); }

  // Return the user key
  Slice user_key() const { return Slice(kstart_, end_ - kstart_ - 8); }
};

总结一下，一个Key中的信息包括：

klength
User Key
Tag
1. Sequence Number
2. Value Type

其中：

Memtable Key
1+2+3
Internal Key
2+3
User Key
2

比较

我们合成了一个包含三个部分的 Internal Key，如果仅仅是这样，直接比较也许还是可行的，毕竟 User Key、Seq、Value Type 啥的都是有层级的。但这些是用 VarInt 存储的，这就不好直接 bitwise 比较了。

Comparator 接口定义在 include/leveldb/comparator.h 里面。包含下面一些成员

int Compare(const Slice& a, const Slice& b) const
比较函数

// Three-way comparison.  Returns value:
//   < 0 iff "a" < "b",
//   == 0 iff "a" == "b",
//   > 0 iff "a" > "b"

void FindShortestSeparator(std::string* start, const Slice& limit)
目的是节约空间。如果 *start 这个字符串小于 limit 字符串，那么就修改 *start，变成一个大小在 *start 和 limit 之间，但是长度最短的字符串。
其方法基于公共前缀，如果 *start 是 helloWorld，limit 是 helloZookeeper，那么旧改 *start 为 helloX，也就是后面的 World 不要了。
这个功能用用在诸如生成 SSTable 的地方。
void FindShortSuccessor(std::string* key)

data 在开头存了对应字符串的长度。所以 GetLengthPrefixedSlice 会先读取长度到 len 里面，这个时候 p 前进指向了实际的字符串，然后创建一个 Slice。

static Slice GetLengthPrefixedSlice(const char* data) {
  uint32_t len;
  const char* p = data;
  p = GetVarint32Ptr(p, p + 5, &len);  // +5: we assume "p" is not corrupted
  return Slice(p, len);
}

MemTable::KeyComparator 就是获取对应的字符串，然后调用 InternalKeyComparator 比较。

int MemTable::KeyComparator::operator()(const char* aptr,
                                        const char* bptr) const {
  // Internal keys are encoded as length-prefixed strings.
  Slice a = GetLengthPrefixedSlice(aptr);
  Slice b = GetLengthPrefixedSlice(bptr);
  return comparator.Compare(a, b);
}

比较的顺序是:

User Key 升序
Sequence Number 降序
这样，会倾向于找新的.

ValueType降序
但考虑到 Sequence Number，大概率用不到。

int InternalKeyComparator::Compare(const Slice& akey, const Slice& bkey) const {
  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;
}

可以看到，在 InternalKeyComparator 里面，还会调用 user_comparator_->Compare 去比较 User Key。它默认是 BytewiseComparator 类型的。

插入

现在有一个 KV 对 key 和 value，都用 Slice 运载的。要把它放到跳表中。
有点类似于 Spark 将 K 和 V 连续放在两个 slot 上，LevelDB 直接将 K 和 V 编码到一起存放。因此可以看到总长度 encoded_len 需要计算下面几部分：

K 的长度，用 VarInt 存
这里 internal_key_size 为啥要加上8呢？这是因为 Sequence Number 和 Type 分别占用了7个 byte 和1个 byte。所以从 User Key 生成 Internal Key 的时候要加上这两部分。
这就是 Memtable Key，相对于 Internal Key 多存储的一块数据。
K 本身
也就是 Internal Key。
V 的长度
V 本身

void MemTable::Add(SequenceNumber s, ValueType type, const Slice& key,
                   const Slice& value) {
  // Format of an entry is concatenation of:
  //  key_size     : varint32 of internal_key.size()
  //  key bytes    : char[internal_key.size()]
  //  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;
...

下面，就是从 Memtable 里面的 Arena 中分配一块内存 buf，然后把上面说的四个字段填进去，最后把 buf 添加到跳表 table_ 里面。跳表插入只需要实现比较操作，这个之前已经定义过了。

...
  char* buf = arena_.Allocate(encoded_len);
  char* p = EncodeVarint32(buf, internal_key_size);
  std::memcpy(p, key.data(), key_size);
  p += key_size;
  EncodeFixed64(p, (s << 8) | type);
  p += 8;
  p = EncodeVarint32(p, val_size);
  std::memcpy(p, value.data(), val_size);
  assert(p + val_size == buf + encoded_len);
  table_.Insert(buf);
}

查找

首先，从 LookupKey 中取得 memtable_key，这个是包含了4个部分的。接着获得跳表的迭代器 iter，定位到第一个大于等于 memkey 的位置。
我们得到 key_ptr 指向 Internal Key，key_length 为 Internal Key 的长度。

bool MemTable::Get(const LookupKey& key, std::string* value, Status* s) {
  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]
    // Check that it belongs to same user key.  We do not check the
    // sequence number since the Seek() call above should have skipped
    // all entries with overly large sequence numbers.
    const char* entry = iter.key();
    uint32_t key_length;
    const char* key_ptr = GetVarint32Ptr(entry, entry + 5, &key_length);
...

接着比较 Value Type，它在 tag 的最后一个字节。注意 tag 的组装方式是 (seq << 8) | type

...
    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;
}

LevelDB之Memtable实现

基本概念

VarInt

Slice

Memtable类

Key

比较

插入

查找

Reference