leveldb笔记之4:MemTable

static uint64_t PackSequenceAndType(uint64_t seq, ValueType t) {
  assert(seq <= kMaxSequenceNumber);
  assert(t <= kValueTypeForSeek);
  return (seq << 8) | t;
}

LookupKey::LookupKey(const Slice& user_key, SequenceNumber s) {
  size_t usize = user_key.size();
  //SequenceNumber + ValueType占8个字节，encode(internal_key_size)至多占5个字节，因此预估至多+13个字节
  size_t needed = usize + 13;  // A conservative estimate
  char* dst;
  if (needed <= sizeof(space_)) {
    dst = space_;
  } else {
    dst = new char[needed];
  }
  //start_指向最开始位置
  start_ = dst;
  //memtable的internal_key_size=usize+8，对应MemTable::Add实现。
  dst = EncodeVarint32(dst, usize + 8);
  //kstart_指向userkey开始位置
  kstart_ = dst;
  memcpy(dst, user_key.data(), usize);
  dst += usize;
  EncodeFixed64(dst, PackSequenceAndType(s, kValueTypeForSeek));
  dst += 8;
  //end_指向(s << 8 | type)的下一个字节
  end_ = dst;
}

int InternalKeyComparator::Compare(const Slice& akey, const Slice& bkey) const {
  // Order by:
  //    increasing user key (according to user-supplied comparator)
  //    decreasing sequence number
  //    decreasing type (though sequence# should be enough to disambiguate)
  //    首先比较userkey，即用户写入的key
  int r = user_comparator_->Compare(ExtractUserKey(akey), ExtractUserKey(bkey));
  // 如果是同一个userkey，继续判断tag是否相等
  if (r == 0) {
    //sequence越大，排序结果越小
    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;
}

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();
  //InternalKey长度 = UserKey长度 + 8bytes(存储SequenceNumber + ValueType)
  size_t internal_key_size = key_size + 8;
  //用户写入的key value在内部实现里增加了sequence type
  //而到了MemTable实际按照以下格式组成一段buffer
  //|encode(internal_key_size)  |key  |sequence  type  |encode(value_size)  |value  |
  //这里先计算大小，申请对应大小的buffer
  const size_t encoded_len =
      VarintLength(internal_key_size) + internal_key_size +
      VarintLength(val_size) + val_size;
  char* buf = arena_.Allocate(encoded_len);
  //append key_size
  char* p = EncodeVarint32(buf, internal_key_size);
  //append key bytes
  memcpy(p, key.data(), key_size);
  p += key_size;
  //append sequence_number && type
  //64bits = 8bytes，前7个bytes存储s，最后一个bytes存储type.
  //这里8bytes对应前面 internal_key_size = key_size + 8
  //也是Fixed64而不是Varint64的原因
  EncodeFixed64(p, (s << 8) | type);
  p += 8;
  //append value_size
  p = EncodeVarint32(p, val_size);
  //append value bytes
  memcpy(p, value.data(), val_size);
  assert(p + val_size == buf + encoded_len);
  //写入table_的buffer包含了key/value及附属信息
  table_.Insert(buf);
}

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;
    //解析出internal_key的长度存储到key_length
    //key_ptr指向internal_key
    const char* key_ptr = GetVarint32Ptr(entry, entry+5, &key_length);
    //Seek返回的是第一个>=key的Node(>= <=> InternalKeyComparator::Compare)
    //因此先判断下userkey是否相等
    if (comparator_.comparator.user_comparator()->Compare(
            Slice(key_ptr, key_length - 8),
            key.user_key()) == 0) {
      // Correct user key
      // tag = (s << 8) | type
      const uint64_t tag = DecodeFixed64(key_ptr + key_length - 8);
      //type存储在最后一个字节
      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;
}