memtable.cc

// Copyright (c) 2011-present, Facebook, Inc. All rights reserved.
// This source code is licensed under both the GPLv2 (found in the
// COPYING file in the root directory) and Apache 2.0 License
// (found in the LICENSE.Apache file in the root directory).
//
// Copyright (c) 2011 The LevelDB Authors. All rights reserved.
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file. See the AUTHORS file for names of contributors.

#include"db/memtable.h"

#include<algorithm>
#include<array>
#include<limits>
#include<memory>
#include<optional>

#include"db/dbformat.h"
#include"db/kv_checksum.h"
#include"db/merge_context.h"
#include"db/merge_helper.h"
#include"db/pinned_iterators_manager.h"
#include"db/range_tombstone_fragmenter.h"
#include"db/read_callback.h"
#include"db/wide/wide_column_serialization.h"
#include"logging/logging.h"
#include"memory/arena.h"
#include"memory/memory_usage.h"
#include"monitoring/perf_context_imp.h"
#include"monitoring/statistics_impl.h"
#include"port/lang.h"
#include"port/port.h"
#include"rocksdb/comparator.h"
#include"rocksdb/env.h"
#include"rocksdb/iterator.h"
#include"rocksdb/merge_operator.h"
#include"rocksdb/slice_transform.h"
#include"rocksdb/types.h"
#include"rocksdb/write_buffer_manager.h"
#include"table/internal_iterator.h"
#include"table/iterator_wrapper.h"
#include"table/merging_iterator.h"
#include"util/autovector.h"
#include"util/coding.h"
#include"util/mutexlock.h"

namespaceROCKSDB_NAMESPACE {

ImmutableMemTableOptions::ImmutableMemTableOptions(
const ImmutableOptions& ioptions,
const MutableCFOptions& mutable_cf_options)
 : arena_block_size(mutable_cf_options.arena_block_size),
memtable_prefix_bloom_bits(
static_cast<uint32_t>(
static_cast<double>(mutable_cf_options.write_buffer_size) *
 mutable_cf_options.memtable_prefix_bloom_size_ratio) *
 8u),
 memtable_huge_page_size(mutable_cf_options.memtable_huge_page_size),
 memtable_whole_key_filtering(
 mutable_cf_options.memtable_whole_key_filtering),
 inplace_update_support(ioptions.inplace_update_support),
 inplace_update_num_locks(mutable_cf_options.inplace_update_num_locks),
 inplace_callback(ioptions.inplace_callback),
 max_successive_merges(mutable_cf_options.max_successive_merges),
 strict_max_successive_merges(
 mutable_cf_options.strict_max_successive_merges),
 statistics(ioptions.stats),
 merge_operator(ioptions.merge_operator.get()),
 info_log(ioptions.logger),
 protection_bytes_per_key(
 mutable_cf_options.memtable_protection_bytes_per_key),
 allow_data_in_errors(ioptions.allow_data_in_errors),
 paranoid_memory_checks(mutable_cf_options.paranoid_memory_checks) {}

MemTable::MemTable(const InternalKeyComparator& cmp,
const ImmutableOptions& ioptions,
const MutableCFOptions& mutable_cf_options,
 WriteBufferManager* write_buffer_manager,
 SequenceNumber latest_seq, uint32_t column_family_id)
 : comparator_(cmp),
 moptions_(ioptions, mutable_cf_options),
 kArenaBlockSize(Arena::OptimizeBlockSize(moptions_.arena_block_size)),
 mem_tracker_(write_buffer_manager),
 arena_(moptions_.arena_block_size,
 (write_buffer_manager != nullptr &&
 (write_buffer_manager->enabled() ||
 write_buffer_manager->cost_to_cache()))
 ? &mem_tracker_
 : nullptr,
 mutable_cf_options.memtable_huge_page_size),
 table_(ioptions.memtable_factory->CreateMemTableRep(
 comparator_, &arena_, mutable_cf_options.prefix_extractor.get(),
 ioptions.logger, column_family_id)),
 range_del_table_(SkipListFactory().CreateMemTableRep(
 comparator_, &arena_, nullptr/* transform */, ioptions.logger,
 column_family_id)),
 is_range_del_table_empty_(true),
 data_size_(0),
 num_entries_(0),
 num_deletes_(0),
 num_range_deletes_(0),
 write_buffer_size_(mutable_cf_options.write_buffer_size),
 first_seqno_(0),
 earliest_seqno_(latest_seq),
 creation_seq_(latest_seq),
 min_prep_log_referenced_(0),
 locks_(moptions_.inplace_update_support
 ? moptions_.inplace_update_num_locks
 : 0),
 prefix_extractor_(mutable_cf_options.prefix_extractor.get()),
 flush_state_(FLUSH_NOT_REQUESTED),
 clock_(ioptions.clock),
 insert_with_hint_prefix_extractor_(
 ioptions.memtable_insert_with_hint_prefix_extractor.get()),
 oldest_key_time_(std::numeric_limits<uint64_t>::max()),
 approximate_memory_usage_(0),
 memtable_max_range_deletions_(
 mutable_cf_options.memtable_max_range_deletions) {
UpdateFlushState();
// something went wrong if we need to flush before inserting anything
assert(!ShouldScheduleFlush());

// use bloom_filter_ for both whole key and prefix bloom filter
if ((prefix_extractor_ || moptions_.memtable_whole_key_filtering) &&
 moptions_.memtable_prefix_bloom_bits > 0) {
 bloom_filter_.reset(
newDynamicBloom(&arena_, moptions_.memtable_prefix_bloom_bits,
6/* hard coded 6 probes */,
 moptions_.memtable_huge_page_size, ioptions.logger));
 }
// Initialize cached_range_tombstone_ here since it could
// be read before it is constructed in MemTable::Add(), which could also lead
// to a data race on the global mutex table backing atomic shared_ptr.
auto new_cache = std::make_shared<FragmentedRangeTombstoneListCache>();
size_t size = cached_range_tombstone_.Size();
for (size_t i = 0; i < size; ++i) {
 std::shared_ptr<FragmentedRangeTombstoneListCache>* local_cache_ref_ptr =
 cached_range_tombstone_.AccessAtCore(i);
auto new_local_cache_ref = std::make_shared<
const std::shared_ptr<FragmentedRangeTombstoneListCache>>(new_cache);
std::atomic_store_explicit(
 local_cache_ref_ptr,
 std::shared_ptr<FragmentedRangeTombstoneListCache>(new_local_cache_ref,
 new_cache.get()),
 std::memory_order_relaxed);
 }
const Comparator* ucmp = cmp.user_comparator();
assert(ucmp);
 ts_sz_ = ucmp->timestamp_size();
}

MemTable::~MemTable() {
 mem_tracker_.FreeMem();
assert(refs_ == 0);
}

size_tMemTable::ApproximateMemoryUsage() {
 autovector<size_t> usages = {
 arena_.ApproximateMemoryUsage(), table_->ApproximateMemoryUsage(),
 range_del_table_->ApproximateMemoryUsage(),
ROCKSDB_NAMESPACE::ApproximateMemoryUsage(insert_hints_)};
size_t total_usage = 0;
for (size_t usage : usages) {
// If usage + total_usage >= kMaxSizet, return kMaxSizet.
// the following variation is to avoid numeric overflow.
if (usage >= std::numeric_limits<size_t>::max() - total_usage) {
return std::numeric_limits<size_t>::max();
 }
 total_usage += usage;
 }
 approximate_memory_usage_.store(total_usage, std::memory_order_relaxed);
// otherwise, return the actual usage
return total_usage;
}

boolMemTable::ShouldFlushNow() {
if (IsMarkedForFlush()) {
// TODO: dedicated flush reason when marked for flush
returntrue;
 }

// This is set if memtable_max_range_deletions is > 0,
// and that many range deletions are done
if (memtable_max_range_deletions_ > 0 &&
 num_range_deletes_.load(std::memory_order_relaxed) >=
static_cast<uint64_t>(memtable_max_range_deletions_)) {
returntrue;
 }

size_t write_buffer_size = write_buffer_size_.load(std::memory_order_relaxed);
// In a lot of times, we cannot allocate arena blocks that exactly matches the
// buffer size. Thus we have to decide if we should over-allocate or
// under-allocate.
// This constant variable can be interpreted as: if we still have more than
// "kAllowOverAllocationRatio * kArenaBlockSize" space left, we'd try to over
// allocate one more block.
constdoublekAllowOverAllocationRatio = 0.6;

// If arena still have room for new block allocation, we can safely say it
// shouldn't flush.
auto allocated_memory = table_->ApproximateMemoryUsage() +
 range_del_table_->ApproximateMemoryUsage() +
 arena_.MemoryAllocatedBytes();

 approximate_memory_usage_.store(allocated_memory, std::memory_order_relaxed);

// if we can still allocate one more block without exceeding the
// over-allocation ratio, then we should not flush.
if (allocated_memory + kArenaBlockSize <
 write_buffer_size + kArenaBlockSize * kAllowOverAllocationRatio) {
returnfalse;
 }

// if user keeps adding entries that exceeds write_buffer_size, we need to
// flush earlier even though we still have much available memory left.
if (allocated_memory >
 write_buffer_size + kArenaBlockSize * kAllowOverAllocationRatio) {
returntrue;
 }

// In this code path, Arena has already allocated its "last block", which
// means the total allocatedmemory size is either:
// (1) "moderately" over allocated the memory (no more than `0.6 * arena
// block size`. Or,
// (2) the allocated memory is less than write buffer size, but we'll stop
// here since if we allocate a new arena block, we'll over allocate too much
// more (half of the arena block size) memory.
//
// In either case, to avoid over-allocate, the last block will stop allocation
// when its usage reaches a certain ratio, which we carefully choose "0.75
// full" as the stop condition because it addresses the following issue with
// great simplicity: What if the next inserted entry's size is
// bigger than AllocatedAndUnused()?
//
// The answer is: if the entry size is also bigger than 0.25 *
// kArenaBlockSize, a dedicated block will be allocated for it; otherwise
// arena will anyway skip the AllocatedAndUnused() and allocate a new, empty
// and regular block. In either case, we *overly* over-allocated.
//
// Therefore, setting the last block to be at most "0.75 full" avoids both
// cases.
//
// NOTE: the average percentage of waste space of this approach can be counted
// as: "arena block size * 0.25 / write buffer size". User who specify a small
// write buffer size and/or big arena block size may suffer.
return arena_.AllocatedAndUnused() < kArenaBlockSize / 4;
}

voidMemTable::UpdateFlushState() {
auto state = flush_state_.load(std::memory_order_relaxed);
if (state == FLUSH_NOT_REQUESTED && ShouldFlushNow()) {
// ignore CAS failure, because that means somebody else requested
// a flush
 flush_state_.compare_exchange_strong(state, FLUSH_REQUESTED,
 std::memory_order_relaxed,
 std::memory_order_relaxed);
 }
}

voidMemTable::UpdateOldestKeyTime() {
uint64_t oldest_key_time = oldest_key_time_.load(std::memory_order_relaxed);
if (oldest_key_time == std::numeric_limits<uint64_t>::max()) {
int64_t current_time = 0;
auto s = clock_->GetCurrentTime(&current_time);
if (s.ok()) {
assert(current_time >= 0);
// If fail, the timestamp is already set.
 oldest_key_time_.compare_exchange_strong(
 oldest_key_time, static_cast<uint64_t>(current_time),
 std::memory_order_relaxed, std::memory_order_relaxed);
 }
 }
}

Status MemTable::VerifyEntryChecksum(constchar* entry,
uint32_t protection_bytes_per_key,
bool allow_data_in_errors) {
if (protection_bytes_per_key == 0) {
returnStatus::OK();
 }
uint32_t key_length;
constchar* key_ptr = GetVarint32Ptr(entry, entry + 5, &key_length);
if (key_ptr == nullptr) {
returnStatus::Corruption("Unable to parse internal key length");
 }
if (key_length < 8) {
returnStatus::Corruption("Memtable entry internal key length too short.");
 }
 Slice user_key = Slice(key_ptr, key_length - 8);

constuint64_t tag = DecodeFixed64(key_ptr + key_length - 8);
 ValueType type;
 SequenceNumber seq;
UnPackSequenceAndType(tag, &seq, &type);

uint32_t value_length = 0;
constchar* value_ptr = GetVarint32Ptr(
 key_ptr + key_length, key_ptr + key_length + 5, &value_length);
if (value_ptr == nullptr) {
returnStatus::Corruption("Unable to parse internal key value");
 }
 Slice value = Slice(value_ptr, value_length);

constchar* checksum_ptr = value_ptr + value_length;
bool match =
ProtectionInfo64()
 .ProtectKVO(user_key, value, type)
 .ProtectS(seq)
 .Verify(static_cast<uint8_t>(protection_bytes_per_key), checksum_ptr);
if (!match) {
 std::string msg(
"Corrupted memtable entry, per key-value checksum verification "
"failed.");
if (allow_data_in_errors) {
 msg.append("Unrecognized value type: " +
std::to_string(static_cast<int>(type)) + ". ");
 msg.append("User key: " + user_key.ToString(/*hex=*/true) + ". ");
 msg.append("seq: " + std::to_string(seq) + ".");
 }
returnStatus::Corruption(msg.c_str());
 }
returnStatus::OK();
}

intMemTable::KeyComparator::operator()(constchar* prefix_len_key1,
constchar* prefix_len_key2) const {
// Internal keys are encoded as length-prefixed strings.
 Slice k1 = GetLengthPrefixedSlice(prefix_len_key1);
 Slice k2 = GetLengthPrefixedSlice(prefix_len_key2);
return comparator.CompareKeySeq(k1, k2);
}

intMemTable::KeyComparator::operator()(
constchar* prefix_len_key, const KeyComparator::DecodedType& key) const {
// Internal keys are encoded as length-prefixed strings.
 Slice a = GetLengthPrefixedSlice(prefix_len_key);
return comparator.CompareKeySeq(a, key);
}

voidMemTableRep::InsertConcurrently(KeyHandle /*handle*/) {
throwstd::runtime_error("concurrent insert not supported");
}

Slice MemTableRep::UserKey(constchar* key) const {
 Slice slice = GetLengthPrefixedSlice(key);
returnSlice(slice.data(), slice.size() - 8);
}

KeyHandle MemTableRep::Allocate(constsize_t len, char** buf) {
 *buf = allocator_->Allocate(len);
returnstatic_cast<KeyHandle>(*buf);
}

// Encode a suitable internal key target for "target" and return it.
// Uses *scratch as scratch space, and the returned pointer will point
// into this scratch space.
constchar* EncodeKey(std::string* scratch, const Slice& target) {
 scratch->clear();
PutVarint32(scratch, static_cast<uint32_t>(target.size()));
 scratch->append(target.data(), target.size());
return scratch->data();
}

classMemTableIterator : publicInternalIterator {
public:
enum Kind { kPointEntries, kRangeDelEntries };
MemTableIterator(
 Kind kind, const MemTable& mem, const ReadOptions& read_options,
 UnownedPtr<const SeqnoToTimeMapping> seqno_to_time_mapping = nullptr,
 Arena* arena = nullptr,
const SliceTransform* cf_prefix_extractor = nullptr)
 : bloom_(nullptr),
prefix_extractor_(mem.prefix_extractor_),
 comparator_(mem.comparator_),
 seqno_to_time_mapping_(seqno_to_time_mapping),
 status_(Status::OK()),
 logger_(mem.moptions_.info_log),
 ts_sz_(mem.ts_sz_),
 protection_bytes_per_key_(mem.moptions_.protection_bytes_per_key),
 valid_(false),
 value_pinned_(
 !mem.GetImmutableMemTableOptions()->inplace_update_support),
 arena_mode_(arena != nullptr),
 paranoid_memory_checks_(mem.moptions_.paranoid_memory_checks),
 allow_data_in_error(mem.moptions_.allow_data_in_errors) {
if (kind == kRangeDelEntries) {
 iter_ = mem.range_del_table_->GetIterator(arena);
 } elseif (prefix_extractor_ != nullptr &&
// NOTE: checking extractor equivalence when not pointer
// equivalent is arguably too expensive for memtable
 prefix_extractor_ == cf_prefix_extractor &&
 (read_options.prefix_same_as_start ||
 (!read_options.total_order_seek &&
 !read_options.auto_prefix_mode))) {
// Auto prefix mode is not implemented in memtable yet.
assert(kind == kPointEntries);
 bloom_ = mem.bloom_filter_.get();
 iter_ = mem.table_->GetDynamicPrefixIterator(arena);
 } else {
assert(kind == kPointEntries);
 iter_ = mem.table_->GetIterator(arena);
 }
 status_.PermitUncheckedError();
 }
// No copying allowed
MemTableIterator(const MemTableIterator&) = delete;
voidoperator=(const MemTableIterator&) = delete;

~MemTableIterator() override {
#ifndef NDEBUG
// Assert that the MemTableIterator is never deleted while
// Pinning is Enabled.
assert(!pinned_iters_mgr_ || !pinned_iters_mgr_->PinningEnabled());
#endif
if (arena_mode_) {
 iter_->~Iterator();
 } else {
delete iter_;
 }
 status_.PermitUncheckedError();
 }

#ifndef NDEBUG
voidSetPinnedItersMgr(PinnedIteratorsManager* pinned_iters_mgr) override {
 pinned_iters_mgr_ = pinned_iters_mgr;
 }
 PinnedIteratorsManager* pinned_iters_mgr_ = nullptr;
#endif

boolValid() constoverride {
// If inner iter_ is not valid, then this iter should also not be valid.
assert(iter_->Valid() || !(valid_ && status_.ok()));
return valid_ && status_.ok();
 }

voidSeek(const Slice& k) override {
PERF_TIMER_GUARD(seek_on_memtable_time);
PERF_COUNTER_ADD(seek_on_memtable_count, 1);
 status_ = Status::OK();
if (bloom_) {
// iterator should only use prefix bloom filter
 Slice user_k_without_ts(ExtractUserKeyAndStripTimestamp(k, ts_sz_));
if (prefix_extractor_->InDomain(user_k_without_ts)) {
 Slice prefix = prefix_extractor_->Transform(user_k_without_ts);
if (!bloom_->MayContain(prefix)) {
PERF_COUNTER_ADD(bloom_memtable_miss_count, 1);
 valid_ = false;
return;
 } else {
PERF_COUNTER_ADD(bloom_memtable_hit_count, 1);
 }
 }
 }
if (paranoid_memory_checks_) {
 status_ = iter_->SeekAndValidate(k, nullptr, allow_data_in_error);
 } else {
 iter_->Seek(k, nullptr);
 }
 valid_ = iter_->Valid();
VerifyEntryChecksum();
 }
voidSeekForPrev(const Slice& k) override {
PERF_TIMER_GUARD(seek_on_memtable_time);
PERF_COUNTER_ADD(seek_on_memtable_count, 1);
 status_ = Status::OK();
if (bloom_) {
 Slice user_k_without_ts(ExtractUserKeyAndStripTimestamp(k, ts_sz_));
if (prefix_extractor_->InDomain(user_k_without_ts)) {
if (!bloom_->MayContain(
 prefix_extractor_->Transform(user_k_without_ts))) {
PERF_COUNTER_ADD(bloom_memtable_miss_count, 1);
 valid_ = false;
return;
 } else {
PERF_COUNTER_ADD(bloom_memtable_hit_count, 1);
 }
 }
 }
if (paranoid_memory_checks_) {
 status_ = iter_->SeekAndValidate(k, nullptr, allow_data_in_error);
 } else {
 iter_->Seek(k, nullptr);
 }
 valid_ = iter_->Valid();
VerifyEntryChecksum();
if (!Valid() && status().ok()) {
SeekToLast();
 }
while (Valid() && comparator_.comparator.Compare(k, key()) < 0) {
Prev();
 }
 }
voidSeekToFirst() override {
 status_ = Status::OK();
 iter_->SeekToFirst();
 valid_ = iter_->Valid();
VerifyEntryChecksum();
 }
voidSeekToLast() override {
 status_ = Status::OK();
 iter_->SeekToLast();
 valid_ = iter_->Valid();
VerifyEntryChecksum();
 }
voidNext() override {
PERF_COUNTER_ADD(next_on_memtable_count, 1);
assert(Valid());
if (paranoid_memory_checks_) {
 status_ = iter_->NextAndValidate(allow_data_in_error);
 } else {
 iter_->Next();
TEST_SYNC_POINT_CALLBACK("MemTableIterator::Next:0", iter_);
 }
 valid_ = iter_->Valid();
VerifyEntryChecksum();
 }
boolNextAndGetResult(IterateResult* result) override {
Next();
bool is_valid = Valid();
if (is_valid) {
 result->key = key();
 result->bound_check_result = IterBoundCheck::kUnknown;
 result->value_prepared = true;
 }
return is_valid;
 }
voidPrev() override {
PERF_COUNTER_ADD(prev_on_memtable_count, 1);
assert(Valid());
if (paranoid_memory_checks_) {
 status_ = iter_->PrevAndValidate(allow_data_in_error);
 } else {
 iter_->Prev();
 }
 valid_ = iter_->Valid();
VerifyEntryChecksum();
 }
 Slice key() constoverride {
assert(Valid());
returnGetLengthPrefixedSlice(iter_->key());
 }

uint64_twrite_unix_time() constoverride {
assert(Valid());
 ParsedInternalKey pikey;
 Status s = ParseInternalKey(key(), &pikey, /*log_err_key=*/false);
if (!s.ok()) {
return std::numeric_limits<uint64_t>::max();
 } elseif (kTypeValuePreferredSeqno == pikey.type) {
returnParsePackedValueForWriteTime(value());
 } elseif (!seqno_to_time_mapping_ || seqno_to_time_mapping_->Empty()) {
return std::numeric_limits<uint64_t>::max();
 }
return seqno_to_time_mapping_->GetProximalTimeBeforeSeqno(pikey.sequence);
 }

 Slice value() constoverride {
assert(Valid());
 Slice key_slice = GetLengthPrefixedSlice(iter_->key());
returnGetLengthPrefixedSlice(key_slice.data() + key_slice.size());
 }

 Status status() constoverride { return status_; }

boolIsKeyPinned() constoverride {
// memtable data is always pinned
returntrue;
 }

boolIsValuePinned() constoverride {
// memtable value is always pinned, except if we allow inplace update.
return value_pinned_;
 }

private:
 DynamicBloom* bloom_;
const SliceTransform* const prefix_extractor_;
const MemTable::KeyComparator comparator_;
 MemTableRep::Iterator* iter_;
// The seqno to time mapping is owned by the SuperVersion.
 UnownedPtr<const SeqnoToTimeMapping> seqno_to_time_mapping_;
 Status status_;
 Logger* logger_;
size_t ts_sz_;
uint32_t protection_bytes_per_key_;
bool valid_;
bool value_pinned_;
bool arena_mode_;
constbool paranoid_memory_checks_;
constbool allow_data_in_error;

voidVerifyEntryChecksum() {
if (protection_bytes_per_key_ > 0 && Valid()) {
 status_ = MemTable::VerifyEntryChecksum(iter_->key(),
 protection_bytes_per_key_);
if (!status_.ok()) {
ROCKS_LOG_ERROR(logger_, "In MemtableIterator: %s", status_.getState());
 }
 }
 }
};

InternalIterator* MemTable::NewIterator(
const ReadOptions& read_options,
 UnownedPtr<const SeqnoToTimeMapping> seqno_to_time_mapping, Arena* arena,
const SliceTransform* prefix_extractor, bool/*for_flush*/) {
assert(arena != nullptr);
auto mem = arena->AllocateAligned(sizeof(MemTableIterator));
returnnew (mem)
MemTableIterator(MemTableIterator::kPointEntries, *this, read_options,
 seqno_to_time_mapping, arena, prefix_extractor);
}

// An iterator wrapper that wraps a MemTableIterator and logically strips each
// key's user-defined timestamp.
classTimestampStrippingIterator : publicInternalIterator {
public:
TimestampStrippingIterator(
 MemTableIterator::Kind kind, const MemTable& memtable,
const ReadOptions& read_options,
 UnownedPtr<const SeqnoToTimeMapping> seqno_to_time_mapping, Arena* arena,
const SliceTransform* cf_prefix_extractor, size_t ts_sz)
 : arena_mode_(arena != nullptr), kind_(kind), ts_sz_(ts_sz) {
assert(ts_sz_ != 0);
void* mem = arena ? arena->AllocateAligned(sizeof(MemTableIterator))
 : operatornew(sizeof(MemTableIterator));
 iter_ = new (mem)
MemTableIterator(kind, memtable, read_options, seqno_to_time_mapping,
 arena, cf_prefix_extractor);
 }

// No copying allowed
TimestampStrippingIterator(const TimestampStrippingIterator&) = delete;
voidoperator=(const TimestampStrippingIterator&) = delete;

~TimestampStrippingIterator() override {
if (arena_mode_) {
 iter_->~MemTableIterator();
 } else {
delete iter_;
 }
 }

voidSetPinnedItersMgr(PinnedIteratorsManager* pinned_iters_mgr) override {
 iter_->SetPinnedItersMgr(pinned_iters_mgr);
 }

boolValid() constoverride { return iter_->Valid(); }
voidSeek(const Slice& k) override {
 iter_->Seek(k);
UpdateKeyAndValueBuffer();
 }
voidSeekForPrev(const Slice& k) override {
 iter_->SeekForPrev(k);
UpdateKeyAndValueBuffer();
 }
voidSeekToFirst() override {
 iter_->SeekToFirst();
UpdateKeyAndValueBuffer();
 }
voidSeekToLast() override {
 iter_->SeekToLast();
UpdateKeyAndValueBuffer();
 }
voidNext() override {
 iter_->Next();
UpdateKeyAndValueBuffer();
 }
boolNextAndGetResult(IterateResult* result) override {
 iter_->Next();
UpdateKeyAndValueBuffer();
bool is_valid = Valid();
if (is_valid) {
 result->key = key();
 result->bound_check_result = IterBoundCheck::kUnknown;
 result->value_prepared = true;
 }
return is_valid;
 }
voidPrev() override {
 iter_->Prev();
UpdateKeyAndValueBuffer();
 }
 Slice key() constoverride {
assert(Valid());
return key_buf_;
 }

uint64_twrite_unix_time() constoverride { return iter_->write_unix_time(); }
 Slice value() constoverride {
if (kind_ == MemTableIterator::Kind::kRangeDelEntries) {
return value_buf_;
 }
return iter_->value();
 }
 Status status() constoverride { return iter_->status(); }
boolIsKeyPinned() constoverride {
// Key is only in a buffer that is updated in each iteration.
returnfalse;
 }
boolIsValuePinned() constoverride {
if (kind_ == MemTableIterator::Kind::kRangeDelEntries) {
returnfalse;
 }
return iter_->IsValuePinned();
 }

private:
voidUpdateKeyAndValueBuffer() {
 key_buf_.clear();
if (kind_ == MemTableIterator::Kind::kRangeDelEntries) {
 value_buf_.clear();
 }
if (!Valid()) {
return;
 }
 Slice original_key = iter_->key();
ReplaceInternalKeyWithMinTimestamp(&key_buf_, original_key, ts_sz_);
if (kind_ == MemTableIterator::Kind::kRangeDelEntries) {
 Slice original_value = iter_->value();
AppendUserKeyWithMinTimestamp(&value_buf_, original_value, ts_sz_);
 }
 }
bool arena_mode_;
 MemTableIterator::Kind kind_;
size_t ts_sz_;
 MemTableIterator* iter_;
 std::string key_buf_;
 std::string value_buf_;
};

InternalIterator* MemTable::NewTimestampStrippingIterator(
const ReadOptions& read_options,
 UnownedPtr<const SeqnoToTimeMapping> seqno_to_time_mapping, Arena* arena,
const SliceTransform* prefix_extractor, size_t ts_sz) {
assert(arena != nullptr);
auto mem = arena->AllocateAligned(sizeof(TimestampStrippingIterator));
returnnew (mem) TimestampStrippingIterator(
 MemTableIterator::kPointEntries, *this, read_options,
 seqno_to_time_mapping, arena, prefix_extractor, ts_sz);
}

FragmentedRangeTombstoneIterator* MemTable::NewRangeTombstoneIterator(
const ReadOptions& read_options, SequenceNumber read_seq,
bool immutable_memtable) {
if (read_options.ignore_range_deletions ||
 is_range_del_table_empty_.load(std::memory_order_relaxed)) {
returnnullptr;
 }
returnNewRangeTombstoneIteratorInternal(read_options, read_seq,
 immutable_memtable);
}

FragmentedRangeTombstoneIterator*
MemTable::NewTimestampStrippingRangeTombstoneIterator(
const ReadOptions& read_options, SequenceNumber read_seq, size_t ts_sz) {
if (read_options.ignore_range_deletions ||
 is_range_del_table_empty_.load(std::memory_order_relaxed)) {
returnnullptr;
 }
if (!timestamp_stripping_fragmented_range_tombstone_list_) {
// TODO: plumb Env::IOActivity, Env::IOPriority
auto* unfragmented_iter = newTimestampStrippingIterator(
 MemTableIterator::kRangeDelEntries, *this, ReadOptions(),
/*seqno_to_time_mapping*/nullptr, /* arena */nullptr,
/* prefix_extractor */nullptr, ts_sz);

 timestamp_stripping_fragmented_range_tombstone_list_ =
 std::make_unique<FragmentedRangeTombstoneList>(
 std::unique_ptr<InternalIterator>(unfragmented_iter),
 comparator_.comparator);
 }
returnnewFragmentedRangeTombstoneIterator(
 timestamp_stripping_fragmented_range_tombstone_list_.get(),
 comparator_.comparator, read_seq, read_options.timestamp);
}

FragmentedRangeTombstoneIterator* MemTable::NewRangeTombstoneIteratorInternal(
const ReadOptions& read_options, SequenceNumber read_seq,
bool immutable_memtable) {
if (immutable_memtable) {
// Note that caller should already have verified that
// !is_range_del_table_empty_
assert(IsFragmentedRangeTombstonesConstructed());
returnnewFragmentedRangeTombstoneIterator(
 fragmented_range_tombstone_list_.get(), comparator_.comparator,
 read_seq, read_options.timestamp);
 }

// takes current cache
 std::shared_ptr<FragmentedRangeTombstoneListCache> cache =
std::atomic_load_explicit(cached_range_tombstone_.Access(),
 std::memory_order_relaxed);
// construct fragmented tombstone list if necessary
if (!cache->initialized.load(std::memory_order_acquire)) {
 cache->reader_mutex.lock();
if (!cache->tombstones) {
auto* unfragmented_iter = newMemTableIterator(
 MemTableIterator::kRangeDelEntries, *this, read_options);
 cache->tombstones.reset(newFragmentedRangeTombstoneList(
 std::unique_ptr<InternalIterator>(unfragmented_iter),
 comparator_.comparator));
 cache->initialized.store(true, std::memory_order_release);
 }
 cache->reader_mutex.unlock();
 }

auto* fragmented_iter = newFragmentedRangeTombstoneIterator(
 cache, comparator_.comparator, read_seq, read_options.timestamp);
return fragmented_iter;
}

voidMemTable::ConstructFragmentedRangeTombstones() {
// There should be no concurrent Construction.
// We could also check fragmented_range_tombstone_list_ to avoid repeate
// constructions. We just construct them here again to be safe.
if (!is_range_del_table_empty_.load(std::memory_order_relaxed)) {
// TODO: plumb Env::IOActivity, Env::IOPriority
auto* unfragmented_iter = newMemTableIterator(
 MemTableIterator::kRangeDelEntries, *this, ReadOptions());

 fragmented_range_tombstone_list_ =
 std::make_unique<FragmentedRangeTombstoneList>(
 std::unique_ptr<InternalIterator>(unfragmented_iter),
 comparator_.comparator);
 }
}

port::RWMutex* MemTable::GetLock(const Slice& key) {
return &locks_[GetSliceRangedNPHash(key, locks_.size())];
}

ReadOnlyMemTable::MemTableStats MemTable::ApproximateStats(
const Slice& start_ikey, const Slice& end_ikey) {
uint64_t entry_count = table_->ApproximateNumEntries(start_ikey, end_ikey);
 entry_count += range_del_table_->ApproximateNumEntries(start_ikey, end_ikey);
if (entry_count == 0) {
return {0, 0};
 }
uint64_t n = num_entries_.load(std::memory_order_relaxed);
if (n == 0) {
return {0, 0};
 }
if (entry_count > n) {
// (range_del_)table_->ApproximateNumEntries() is just an estimate so it can
// be larger than actual entries we have. Cap it to entries we have to limit
// the inaccuracy.
 entry_count = n;
 }
uint64_t data_size = data_size_.load(std::memory_order_relaxed);
return {entry_count * (data_size / n), entry_count};
}

Status MemTable::VerifyEncodedEntry(Slice encoded,
const ProtectionInfoKVOS64& kv_prot_info) {
uint32_t ikey_len = 0;
if (!GetVarint32(&encoded, &ikey_len)) {
returnStatus::Corruption("Unable to parse internal key length");
 }
if (ikey_len < 8 + ts_sz_) {
returnStatus::Corruption("Internal key length too short");
 }
if (ikey_len > encoded.size()) {
returnStatus::Corruption("Internal key length too long");
 }
uint32_t value_len = 0;
constsize_t user_key_len = ikey_len - 8;
 Slice key(encoded.data(), user_key_len);
 encoded.remove_prefix(user_key_len);

uint64_t packed = DecodeFixed64(encoded.data());
 ValueType value_type = kMaxValue;
 SequenceNumber sequence_number = kMaxSequenceNumber;
UnPackSequenceAndType(packed, &sequence_number, &value_type);
 encoded.remove_prefix(8);

if (!GetVarint32(&encoded, &value_len)) {
returnStatus::Corruption("Unable to parse value length");
 }
if (value_len < encoded.size()) {
returnStatus::Corruption("Value length too short");
 }
if (value_len > encoded.size()) {
returnStatus::Corruption("Value length too long");
 }
 Slice value(encoded.data(), value_len);

return kv_prot_info.StripS(sequence_number)
 .StripKVO(key, value, value_type)
 .GetStatus();
}

voidMemTable::UpdateEntryChecksum(const ProtectionInfoKVOS64* kv_prot_info,
const Slice& key, const Slice& value,
 ValueType type, SequenceNumber s,
char* checksum_ptr) {
if (moptions_.protection_bytes_per_key == 0) {
return;
 }

if (kv_prot_info == nullptr) {
ProtectionInfo64()
 .ProtectKVO(key, value, type)
 .ProtectS(s)
 .Encode(static_cast<uint8_t>(moptions_.protection_bytes_per_key),
 checksum_ptr);
 } else {
 kv_prot_info->Encode(
static_cast<uint8_t>(moptions_.protection_bytes_per_key), checksum_ptr);
 }
}

Status MemTable::Add(SequenceNumber s, ValueType type,
const Slice& key, /* user key */
const Slice& value,
const ProtectionInfoKVOS64* kv_prot_info,
bool allow_concurrent,
 MemTablePostProcessInfo* post_process_info, void** hint) {
// 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()]
// checksum : char[moptions_.protection_bytes_per_key]
uint32_t key_size = static_cast<uint32_t>(key.size());
uint32_t val_size = static_cast<uint32_t>(value.size());
uint32_t internal_key_size = key_size + 8;
constuint32_t encoded_len = VarintLength(internal_key_size) +
 internal_key_size + VarintLength(val_size) +
 val_size + moptions_.protection_bytes_per_key;
char* buf = nullptr;
 std::unique_ptr<MemTableRep>& table =
 type == kTypeRangeDeletion ? range_del_table_ : table_;
 KeyHandle handle = table->Allocate(encoded_len, &buf);

char* p = EncodeVarint32(buf, internal_key_size);
memcpy(p, key.data(), key_size);
 Slice key_slice(p, key_size);
 p += key_size;
uint64_t packed = PackSequenceAndType(s, type);
EncodeFixed64(p, packed);
 p += 8;
 p = EncodeVarint32(p, val_size);
memcpy(p, value.data(), val_size);
assert((unsigned)(p + val_size - buf + moptions_.protection_bytes_per_key) ==
 (unsigned)encoded_len);

UpdateEntryChecksum(kv_prot_info, key, value, type, s,
 buf + encoded_len - moptions_.protection_bytes_per_key);
 Slice encoded(buf, encoded_len - moptions_.protection_bytes_per_key);
if (kv_prot_info != nullptr) {
TEST_SYNC_POINT_CALLBACK("MemTable::Add:Encoded", &encoded);
 Status status = VerifyEncodedEntry(encoded, *kv_prot_info);
if (!status.ok()) {
return status;
 }
 }

 Slice key_without_ts = StripTimestampFromUserKey(key, ts_sz_);

if (!allow_concurrent) {
// Extract prefix for insert with hint. Hints are for point key table
// (`table_`) only, not `range_del_table_`.
if (table == table_ && insert_with_hint_prefix_extractor_ != nullptr &&
 insert_with_hint_prefix_extractor_->InDomain(key_slice)) {
 Slice prefix = insert_with_hint_prefix_extractor_->Transform(key_slice);
bool res = table->InsertKeyWithHint(handle, &insert_hints_[prefix]);
if (UNLIKELY(!res)) {
returnStatus::TryAgain("key+seq exists");
 }
 } else {
bool res = table->InsertKey(handle);
if (UNLIKELY(!res)) {
returnStatus::TryAgain("key+seq exists");
 }
 }

// this is a bit ugly, but is the way to avoid locked instructions
// when incrementing an atomic
 num_entries_.store(num_entries_.load(std::memory_order_relaxed) + 1,
 std::memory_order_relaxed);
 data_size_.store(data_size_.load(std::memory_order_relaxed) + encoded_len,
 std::memory_order_relaxed);
if (type == kTypeDeletion || type == kTypeSingleDeletion ||
 type == kTypeDeletionWithTimestamp) {
 num_deletes_.store(num_deletes_.load(std::memory_order_relaxed) + 1,
 std::memory_order_relaxed);
 } elseif (type == kTypeRangeDeletion) {
uint64_t val = num_range_deletes_.load(std::memory_order_relaxed) + 1;
 num_range_deletes_.store(val, std::memory_order_relaxed);
 }

if (bloom_filter_ && prefix_extractor_ &&
 prefix_extractor_->InDomain(key_without_ts)) {
 bloom_filter_->Add(prefix_extractor_->Transform(key_without_ts));
 }
if (bloom_filter_ && moptions_.memtable_whole_key_filtering) {
 bloom_filter_->Add(key_without_ts);
 }