rocksdb/memtable/wbwi_memtable.cc

148 lines
5.8 KiB
C++

// Copyright (c) Meta Platforms, Inc. and affiliates.
// 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).
#include "memtable/wbwi_memtable.h"
#include "db/memtable.h"
namespace ROCKSDB_NAMESPACE {
const std::unordered_map<WriteType, ValueType>
WBWIMemTableIterator::WriteTypeToValueTypeMap = {
{kPutRecord, kTypeValue},
{kMergeRecord, kTypeMerge},
{kDeleteRecord, kTypeDeletion},
{kSingleDeleteRecord, kTypeSingleDeletion},
{kDeleteRangeRecord, kTypeRangeDeletion},
{kPutEntityRecord, kTypeWideColumnEntity},
// Only the above record types are added to WBWI.
// kLogDataRecord, kXIDRecord, kUnknownRecord
};
bool WBWIMemTable::Get(const LookupKey& key, std::string* value,
PinnableWideColumns* columns, std::string* timestamp,
Status* s, MergeContext* merge_context,
SequenceNumber* max_covering_tombstone_seq,
SequenceNumber* out_seq, const ReadOptions&,
bool immutable_memtable, ReadCallback* callback,
bool* is_blob_index, bool do_merge) {
(void)immutable_memtable;
(void)timestamp;
(void)columns;
assert(immutable_memtable);
assert(!timestamp); // TODO: support UDT
assert(!columns); // TODO: support WideColumn
assert(global_seqno_ != kMaxSequenceNumber);
// WBWI does not support DeleteRange yet.
assert(!wbwi_->GetWriteBatch()->HasDeleteRange());
[[maybe_unused]] SequenceNumber read_seq =
GetInternalKeySeqno(key.internal_key());
std::unique_ptr<InternalIterator> iter{NewIterator()};
iter->Seek(key.internal_key());
const Slice lookup_user_key = key.user_key();
while (iter->Valid() && comparator_->EqualWithoutTimestamp(
ExtractUserKey(iter->key()), lookup_user_key)) {
uint64_t tag = ExtractInternalKeyFooter(iter->key());
ValueType type;
SequenceNumber seq;
UnPackSequenceAndType(tag, &seq, &type);
// Unsupported operations.
assert(type != kTypeBlobIndex);
assert(type != kTypeWideColumnEntity);
assert(type != kTypeValuePreferredSeqno);
assert(type != kTypeDeletionWithTimestamp);
assert(type != kTypeMerge);
if (!callback || callback->IsVisible(seq)) {
if (*out_seq == kMaxSequenceNumber) {
*out_seq = std::max(seq, *max_covering_tombstone_seq);
}
if (*max_covering_tombstone_seq > seq) {
type = kTypeRangeDeletion;
}
switch (type) {
case kTypeValue: {
HandleTypeValue(lookup_user_key, iter->value(), iter->IsValuePinned(),
do_merge, s->IsMergeInProgress(), merge_context,
moptions_.merge_operator, clock_,
moptions_.statistics, moptions_.info_log, s, value,
columns, is_blob_index);
assert(seq <= read_seq);
return /*found_final_value=*/true;
}
case kTypeDeletion:
case kTypeSingleDeletion:
case kTypeRangeDeletion: {
HandleTypeDeletion(lookup_user_key, s->IsMergeInProgress(),
merge_context, moptions_.merge_operator, clock_,
moptions_.statistics, moptions_.info_log, s, value,
columns);
assert(seq <= read_seq);
return /*found_final_value=*/true;
}
default: {
std::string msg("Unrecognized or unsupported value type: " +
std::to_string(static_cast<int>(type)) + ". ");
msg.append("User key: " +
ExtractUserKey(iter->key()).ToString(/*hex=*/true) + ". ");
msg.append("seq: " + std::to_string(seq) + ".");
*s = Status::Corruption(msg.c_str());
return /*found_final_value=*/true;
}
}
}
// Current key not visible or we read a merge key
assert(s->IsMergeInProgress() || (callback && !callback->IsVisible(seq)));
iter->Next();
}
if (!iter->status().ok() &&
(s->ok() || s->IsMergeInProgress() || s->IsNotFound())) {
*s = iter->status();
// stop further look up
return true;
}
return /*found_final_value=*/false;
}
void WBWIMemTable::MultiGet(const ReadOptions& read_options,
MultiGetRange* range, ReadCallback* callback,
bool immutable_memtable) {
(void)immutable_memtable;
// Should only be used as immutable memtable.
assert(immutable_memtable);
// TODO: reuse the InternalIterator created in Get().
for (auto iter = range->begin(); iter != range->end(); ++iter) {
SequenceNumber dummy_seq = 0;
bool found_final_value =
Get(*iter->lkey, iter->value ? iter->value->GetSelf() : nullptr,
iter->columns, iter->timestamp, iter->s, &(iter->merge_context),
&(iter->max_covering_tombstone_seq), &dummy_seq, read_options, true,
callback, nullptr, true);
if (found_final_value) {
if (iter->s->ok() || iter->s->IsNotFound()) {
if (iter->value) {
iter->value->PinSelf();
range->AddValueSize(iter->value->size());
} else {
assert(iter->columns);
range->AddValueSize(iter->columns->serialized_size());
}
}
range->MarkKeyDone(iter);
if (range->GetValueSize() > read_options.value_size_soft_limit) {
// Set all remaining keys in range to Abort
for (auto range_iter = range->begin(); range_iter != range->end();
++range_iter) {
range->MarkKeyDone(range_iter);
*(range_iter->s) = Status::Aborted();
}
break;
}
}
}
}
} // namespace ROCKSDB_NAMESPACE