mirror of https://github.com/facebook/rocksdb.git
Add support for wide-column point lookups (#10540)
Summary: The patch adds a new API `GetEntity` that can be used to perform wide-column point lookups. It also extends the `Get` code path and the `MemTable` / `MemTableList` and `Version` / `GetContext` logic accordingly so that wide-column entities can be served from both memtables and SSTs. If the result of a lookup is a wide-column entity (`kTypeWideColumnEntity`), it is passed to the application in deserialized form; if it is a plain old key-value (`kTypeValue`), it is presented as a wide-column entity with a single default (anonymous) column. (In contrast, regular `Get` returns plain old key-values as-is, and returns the value of the default column for wide-column entities, see https://github.com/facebook/rocksdb/issues/10483 .) The result of `GetEntity` is a self-contained `PinnableWideColumns` object. `PinnableWideColumns` contains a `PinnableSlice`, which either stores the underlying data in its own buffer or holds on to a cache handle. It also contains a `WideColumns` instance, which indexes the contents of the `PinnableSlice`, so applications can access the values of columns efficiently. There are several pieces of functionality which are currently not supported for wide-column entities: there is currently no `MultiGetEntity` or wide-column iterator; also, `Merge` and `GetMergeOperands` are not supported, and there is no `GetEntity` implementation for read-only and secondary instances. We plan to implement these in future PRs. Pull Request resolved: https://github.com/facebook/rocksdb/pull/10540 Test Plan: `make check` Reviewed By: akankshamahajan15 Differential Revision: D38847474 Pulled By: ltamasi fbshipit-source-id: 42311a34ccdfe88b3775e847a5e2a5296e002b5b
This commit is contained in:
parent
2553d1efa1
commit
81388b36e0
|
@ -684,6 +684,7 @@ set(SOURCES
|
|||
db/wal_edit.cc
|
||||
db/wal_manager.cc
|
||||
db/wide/wide_column_serialization.cc
|
||||
db/wide/wide_columns.cc
|
||||
db/write_batch.cc
|
||||
db/write_batch_base.cc
|
||||
db/write_controller.cc
|
||||
|
@ -1510,7 +1511,7 @@ if(WITH_BENCHMARK_TOOLS)
|
|||
endif()
|
||||
|
||||
option(WITH_TRACE_TOOLS "build with trace tools" ON)
|
||||
if(WITH_TRACE_TOOLS)
|
||||
if(WITH_TRACE_TOOLS)
|
||||
add_executable(block_cache_trace_analyzer_tool${ARTIFACT_SUFFIX}
|
||||
tools/block_cache_analyzer/block_cache_trace_analyzer_tool.cc)
|
||||
target_link_libraries(block_cache_trace_analyzer_tool${ARTIFACT_SUFFIX}
|
||||
|
|
2
TARGETS
2
TARGETS
|
@ -99,6 +99,7 @@ cpp_library_wrapper(name="rocksdb_lib", srcs=[
|
|||
"db/wal_edit.cc",
|
||||
"db/wal_manager.cc",
|
||||
"db/wide/wide_column_serialization.cc",
|
||||
"db/wide/wide_columns.cc",
|
||||
"db/write_batch.cc",
|
||||
"db/write_batch_base.cc",
|
||||
"db/write_controller.cc",
|
||||
|
@ -435,6 +436,7 @@ cpp_library_wrapper(name="rocksdb_whole_archive_lib", srcs=[
|
|||
"db/wal_edit.cc",
|
||||
"db/wal_manager.cc",
|
||||
"db/wide/wide_column_serialization.cc",
|
||||
"db/wide/wide_columns.cc",
|
||||
"db/write_batch.cc",
|
||||
"db/write_batch_base.cc",
|
||||
"db/write_controller.cc",
|
||||
|
|
|
@ -72,9 +72,9 @@ Status CompactedDBImpl::Get(const ReadOptions& options, ColumnFamilyHandle*,
|
|||
user_comparator_->timestamp_size() > 0 ? timestamp : nullptr;
|
||||
LookupKey lkey(key, kMaxSequenceNumber, options.timestamp);
|
||||
GetContext get_context(user_comparator_, nullptr, nullptr, nullptr,
|
||||
GetContext::kNotFound, lkey.user_key(), value, ts,
|
||||
nullptr, nullptr, true, nullptr, nullptr, nullptr,
|
||||
nullptr, &read_cb);
|
||||
GetContext::kNotFound, lkey.user_key(), value,
|
||||
/*columns=*/nullptr, ts, nullptr, nullptr, true,
|
||||
nullptr, nullptr, nullptr, nullptr, &read_cb);
|
||||
|
||||
const FdWithKeyRange& f = files_.files[FindFile(lkey.user_key())];
|
||||
if (user_comparator_->CompareWithoutTimestamp(
|
||||
|
@ -159,7 +159,7 @@ std::vector<Status> CompactedDBImpl::MultiGet(
|
|||
std::string* timestamp = timestamps ? &(*timestamps)[idx] : nullptr;
|
||||
GetContext get_context(
|
||||
user_comparator_, nullptr, nullptr, nullptr, GetContext::kNotFound,
|
||||
lkey.user_key(), &pinnable_val,
|
||||
lkey.user_key(), &pinnable_val, /*columns=*/nullptr,
|
||||
user_comparator_->timestamp_size() > 0 ? timestamp : nullptr, nullptr,
|
||||
nullptr, true, nullptr, nullptr, nullptr, nullptr, &read_cb);
|
||||
Status s = r->Get(options, lkey.internal_key(), &get_context, nullptr);
|
||||
|
|
|
@ -1822,6 +1822,28 @@ Status DBImpl::Get(const ReadOptions& read_options,
|
|||
return s;
|
||||
}
|
||||
|
||||
Status DBImpl::GetEntity(const ReadOptions& read_options,
|
||||
ColumnFamilyHandle* column_family, const Slice& key,
|
||||
PinnableWideColumns* columns) {
|
||||
if (!column_family) {
|
||||
return Status::InvalidArgument(
|
||||
"Cannot call GetEntity without a column family handle");
|
||||
}
|
||||
|
||||
if (!columns) {
|
||||
return Status::InvalidArgument(
|
||||
"Cannot call GetEntity without a PinnableWideColumns object");
|
||||
}
|
||||
|
||||
columns->Reset();
|
||||
|
||||
GetImplOptions get_impl_options;
|
||||
get_impl_options.column_family = column_family;
|
||||
get_impl_options.columns = columns;
|
||||
|
||||
return GetImpl(read_options, key, get_impl_options);
|
||||
}
|
||||
|
||||
bool DBImpl::ShouldReferenceSuperVersion(const MergeContext& merge_context) {
|
||||
// If both thresholds are reached, a function returning merge operands as
|
||||
// `PinnableSlice`s should reference the `SuperVersion` to avoid large and/or
|
||||
|
@ -1853,7 +1875,8 @@ bool DBImpl::ShouldReferenceSuperVersion(const MergeContext& merge_context) {
|
|||
Status DBImpl::GetImpl(const ReadOptions& read_options, const Slice& key,
|
||||
GetImplOptions& get_impl_options) {
|
||||
assert(get_impl_options.value != nullptr ||
|
||||
get_impl_options.merge_operands != nullptr);
|
||||
get_impl_options.merge_operands != nullptr ||
|
||||
get_impl_options.columns != nullptr);
|
||||
|
||||
assert(get_impl_options.column_family);
|
||||
|
||||
|
@ -1980,31 +2003,46 @@ Status DBImpl::GetImpl(const ReadOptions& read_options, const Slice& key,
|
|||
if (!skip_memtable) {
|
||||
// Get value associated with key
|
||||
if (get_impl_options.get_value) {
|
||||
if (sv->mem->Get(lkey, get_impl_options.value->GetSelf(), timestamp, &s,
|
||||
&merge_context, &max_covering_tombstone_seq,
|
||||
read_options, false /* immutable_memtable */,
|
||||
get_impl_options.callback,
|
||||
get_impl_options.is_blob_index)) {
|
||||
if (sv->mem->Get(
|
||||
lkey,
|
||||
get_impl_options.value ? get_impl_options.value->GetSelf()
|
||||
: nullptr,
|
||||
get_impl_options.columns, timestamp, &s, &merge_context,
|
||||
&max_covering_tombstone_seq, read_options,
|
||||
false /* immutable_memtable */, get_impl_options.callback,
|
||||
get_impl_options.is_blob_index)) {
|
||||
done = true;
|
||||
get_impl_options.value->PinSelf();
|
||||
|
||||
if (get_impl_options.value) {
|
||||
get_impl_options.value->PinSelf();
|
||||
}
|
||||
|
||||
RecordTick(stats_, MEMTABLE_HIT);
|
||||
} else if ((s.ok() || s.IsMergeInProgress()) &&
|
||||
sv->imm->Get(lkey, get_impl_options.value->GetSelf(),
|
||||
timestamp, &s, &merge_context,
|
||||
&max_covering_tombstone_seq, read_options,
|
||||
get_impl_options.callback,
|
||||
sv->imm->Get(lkey,
|
||||
get_impl_options.value
|
||||
? get_impl_options.value->GetSelf()
|
||||
: nullptr,
|
||||
get_impl_options.columns, timestamp, &s,
|
||||
&merge_context, &max_covering_tombstone_seq,
|
||||
read_options, get_impl_options.callback,
|
||||
get_impl_options.is_blob_index)) {
|
||||
done = true;
|
||||
get_impl_options.value->PinSelf();
|
||||
|
||||
if (get_impl_options.value) {
|
||||
get_impl_options.value->PinSelf();
|
||||
}
|
||||
|
||||
RecordTick(stats_, MEMTABLE_HIT);
|
||||
}
|
||||
} else {
|
||||
// Get Merge Operands associated with key, Merge Operands should not be
|
||||
// merged and raw values should be returned to the user.
|
||||
if (sv->mem->Get(lkey, /*value*/ nullptr, /*timestamp=*/nullptr, &s,
|
||||
&merge_context, &max_covering_tombstone_seq,
|
||||
read_options, false /* immutable_memtable */, nullptr,
|
||||
nullptr, false)) {
|
||||
if (sv->mem->Get(lkey, /*value=*/nullptr, /*columns=*/nullptr,
|
||||
/*timestamp=*/nullptr, &s, &merge_context,
|
||||
&max_covering_tombstone_seq, read_options,
|
||||
false /* immutable_memtable */, nullptr, nullptr,
|
||||
false)) {
|
||||
done = true;
|
||||
RecordTick(stats_, MEMTABLE_HIT);
|
||||
} else if ((s.ok() || s.IsMergeInProgress()) &&
|
||||
|
@ -2026,8 +2064,9 @@ Status DBImpl::GetImpl(const ReadOptions& read_options, const Slice& key,
|
|||
if (!done) {
|
||||
PERF_TIMER_GUARD(get_from_output_files_time);
|
||||
sv->current->Get(
|
||||
read_options, lkey, get_impl_options.value, timestamp, &s,
|
||||
&merge_context, &max_covering_tombstone_seq, &pinned_iters_mgr,
|
||||
read_options, lkey, get_impl_options.value, get_impl_options.columns,
|
||||
timestamp, &s, &merge_context, &max_covering_tombstone_seq,
|
||||
&pinned_iters_mgr,
|
||||
get_impl_options.get_value ? get_impl_options.value_found : nullptr,
|
||||
nullptr, nullptr,
|
||||
get_impl_options.get_value ? get_impl_options.callback : nullptr,
|
||||
|
@ -2043,7 +2082,11 @@ Status DBImpl::GetImpl(const ReadOptions& read_options, const Slice& key,
|
|||
size_t size = 0;
|
||||
if (s.ok()) {
|
||||
if (get_impl_options.get_value) {
|
||||
size = get_impl_options.value->size();
|
||||
if (get_impl_options.value) {
|
||||
size = get_impl_options.value->size();
|
||||
} else if (get_impl_options.columns) {
|
||||
size = get_impl_options.columns->serialized_size();
|
||||
}
|
||||
} else {
|
||||
// Return all merge operands for get_impl_options.key
|
||||
*get_impl_options.number_of_operands =
|
||||
|
@ -2252,14 +2295,14 @@ std::vector<Status> DBImpl::MultiGet(
|
|||
has_unpersisted_data_.load(std::memory_order_relaxed));
|
||||
bool done = false;
|
||||
if (!skip_memtable) {
|
||||
if (super_version->mem->Get(lkey, value, timestamp, &s, &merge_context,
|
||||
&max_covering_tombstone_seq, read_options,
|
||||
false /* immutable_memtable */,
|
||||
read_callback)) {
|
||||
if (super_version->mem->Get(
|
||||
lkey, value, /*columns=*/nullptr, timestamp, &s, &merge_context,
|
||||
&max_covering_tombstone_seq, read_options,
|
||||
false /* immutable_memtable */, read_callback)) {
|
||||
done = true;
|
||||
RecordTick(stats_, MEMTABLE_HIT);
|
||||
} else if (super_version->imm->Get(lkey, value, timestamp, &s,
|
||||
&merge_context,
|
||||
} else if (super_version->imm->Get(lkey, value, /*columns=*/nullptr,
|
||||
timestamp, &s, &merge_context,
|
||||
&max_covering_tombstone_seq,
|
||||
read_options, read_callback)) {
|
||||
done = true;
|
||||
|
@ -2270,9 +2313,9 @@ std::vector<Status> DBImpl::MultiGet(
|
|||
PinnableSlice pinnable_val;
|
||||
PERF_TIMER_GUARD(get_from_output_files_time);
|
||||
PinnedIteratorsManager pinned_iters_mgr;
|
||||
super_version->current->Get(read_options, lkey, &pinnable_val, timestamp,
|
||||
&s, &merge_context,
|
||||
&max_covering_tombstone_seq,
|
||||
super_version->current->Get(read_options, lkey, &pinnable_val,
|
||||
/*columns=*/nullptr, timestamp, &s,
|
||||
&merge_context, &max_covering_tombstone_seq,
|
||||
&pinned_iters_mgr, /*value_found=*/nullptr,
|
||||
/*key_exists=*/nullptr,
|
||||
/*seq=*/nullptr, read_callback);
|
||||
|
@ -4861,8 +4904,8 @@ Status DBImpl::GetLatestSequenceForKey(
|
|||
*found_record_for_key = false;
|
||||
|
||||
// Check if there is a record for this key in the latest memtable
|
||||
sv->mem->Get(lkey, /*value=*/nullptr, timestamp, &s, &merge_context,
|
||||
&max_covering_tombstone_seq, seq, read_options,
|
||||
sv->mem->Get(lkey, /*value=*/nullptr, /*columns=*/nullptr, timestamp, &s,
|
||||
&merge_context, &max_covering_tombstone_seq, seq, read_options,
|
||||
false /* immutable_memtable */, nullptr /*read_callback*/,
|
||||
is_blob_index);
|
||||
|
||||
|
@ -4895,8 +4938,8 @@ Status DBImpl::GetLatestSequenceForKey(
|
|||
}
|
||||
|
||||
// Check if there is a record for this key in the immutable memtables
|
||||
sv->imm->Get(lkey, /*value=*/nullptr, timestamp, &s, &merge_context,
|
||||
&max_covering_tombstone_seq, seq, read_options,
|
||||
sv->imm->Get(lkey, /*value=*/nullptr, /*columns=*/nullptr, timestamp, &s,
|
||||
&merge_context, &max_covering_tombstone_seq, seq, read_options,
|
||||
nullptr /*read_callback*/, is_blob_index);
|
||||
|
||||
if (!(s.ok() || s.IsNotFound() || s.IsMergeInProgress())) {
|
||||
|
@ -4927,9 +4970,10 @@ Status DBImpl::GetLatestSequenceForKey(
|
|||
}
|
||||
|
||||
// Check if there is a record for this key in the immutable memtables
|
||||
sv->imm->GetFromHistory(lkey, /*value=*/nullptr, timestamp, &s,
|
||||
&merge_context, &max_covering_tombstone_seq, seq,
|
||||
read_options, is_blob_index);
|
||||
sv->imm->GetFromHistory(lkey, /*value=*/nullptr, /*columns=*/nullptr,
|
||||
timestamp, &s, &merge_context,
|
||||
&max_covering_tombstone_seq, seq, read_options,
|
||||
is_blob_index);
|
||||
|
||||
if (!(s.ok() || s.IsNotFound() || s.IsMergeInProgress())) {
|
||||
// unexpected error reading memtable.
|
||||
|
@ -4962,8 +5006,8 @@ Status DBImpl::GetLatestSequenceForKey(
|
|||
if (!cache_only) {
|
||||
// Check tables
|
||||
PinnedIteratorsManager pinned_iters_mgr;
|
||||
sv->current->Get(read_options, lkey, /*value=*/nullptr, timestamp, &s,
|
||||
&merge_context, &max_covering_tombstone_seq,
|
||||
sv->current->Get(read_options, lkey, /*value=*/nullptr, /*columns=*/nullptr,
|
||||
timestamp, &s, &merge_context, &max_covering_tombstone_seq,
|
||||
&pinned_iters_mgr, nullptr /* value_found */,
|
||||
found_record_for_key, seq, nullptr /*read_callback*/,
|
||||
is_blob_index);
|
||||
|
|
|
@ -238,6 +238,11 @@ class DBImpl : public DB {
|
|||
ColumnFamilyHandle* column_family, const Slice& key,
|
||||
PinnableSlice* value, std::string* timestamp) override;
|
||||
|
||||
using DB::GetEntity;
|
||||
Status GetEntity(const ReadOptions& options,
|
||||
ColumnFamilyHandle* column_family, const Slice& key,
|
||||
PinnableWideColumns* columns) override;
|
||||
|
||||
using DB::GetMergeOperands;
|
||||
Status GetMergeOperands(const ReadOptions& options,
|
||||
ColumnFamilyHandle* column_family, const Slice& key,
|
||||
|
@ -592,6 +597,7 @@ class DBImpl : public DB {
|
|||
struct GetImplOptions {
|
||||
ColumnFamilyHandle* column_family = nullptr;
|
||||
PinnableSlice* value = nullptr;
|
||||
PinnableWideColumns* columns = nullptr;
|
||||
std::string* timestamp = nullptr;
|
||||
bool* value_found = nullptr;
|
||||
ReadCallback* callback = nullptr;
|
||||
|
|
|
@ -85,18 +85,18 @@ Status DBImplReadOnly::Get(const ReadOptions& read_options,
|
|||
SequenceNumber max_covering_tombstone_seq = 0;
|
||||
LookupKey lkey(key, snapshot, read_options.timestamp);
|
||||
PERF_TIMER_STOP(get_snapshot_time);
|
||||
if (super_version->mem->Get(lkey, pinnable_val->GetSelf(), ts, &s,
|
||||
&merge_context, &max_covering_tombstone_seq,
|
||||
read_options, false /* immutable_memtable */,
|
||||
&read_cb)) {
|
||||
if (super_version->mem->Get(lkey, pinnable_val->GetSelf(),
|
||||
/*columns=*/nullptr, ts, &s, &merge_context,
|
||||
&max_covering_tombstone_seq, read_options,
|
||||
false /* immutable_memtable */, &read_cb)) {
|
||||
pinnable_val->PinSelf();
|
||||
RecordTick(stats_, MEMTABLE_HIT);
|
||||
} else {
|
||||
PERF_TIMER_GUARD(get_from_output_files_time);
|
||||
PinnedIteratorsManager pinned_iters_mgr;
|
||||
super_version->current->Get(
|
||||
read_options, lkey, pinnable_val, ts, &s, &merge_context,
|
||||
&max_covering_tombstone_seq, &pinned_iters_mgr,
|
||||
read_options, lkey, pinnable_val, /*columns=*/nullptr, ts, &s,
|
||||
&merge_context, &max_covering_tombstone_seq, &pinned_iters_mgr,
|
||||
/*value_found*/ nullptr,
|
||||
/*key_exists*/ nullptr, /*seq*/ nullptr, &read_cb,
|
||||
/*is_blob*/ nullptr,
|
||||
|
|
|
@ -390,17 +390,18 @@ Status DBImplSecondary::GetImpl(const ReadOptions& read_options,
|
|||
const Comparator* ucmp = column_family->GetComparator();
|
||||
assert(ucmp);
|
||||
std::string* ts = ucmp->timestamp_size() > 0 ? timestamp : nullptr;
|
||||
if (super_version->mem->Get(lkey, pinnable_val->GetSelf(), ts, &s,
|
||||
&merge_context, &max_covering_tombstone_seq,
|
||||
read_options, false /* immutable_memtable */,
|
||||
&read_cb)) {
|
||||
if (super_version->mem->Get(lkey, pinnable_val->GetSelf(),
|
||||
/*columns=*/nullptr, ts, &s, &merge_context,
|
||||
&max_covering_tombstone_seq, read_options,
|
||||
false /* immutable_memtable */, &read_cb)) {
|
||||
done = true;
|
||||
pinnable_val->PinSelf();
|
||||
RecordTick(stats_, MEMTABLE_HIT);
|
||||
} else if ((s.ok() || s.IsMergeInProgress()) &&
|
||||
super_version->imm->Get(
|
||||
lkey, pinnable_val->GetSelf(), ts, &s, &merge_context,
|
||||
&max_covering_tombstone_seq, read_options, &read_cb)) {
|
||||
lkey, pinnable_val->GetSelf(), /*columns=*/nullptr, ts, &s,
|
||||
&merge_context, &max_covering_tombstone_seq, read_options,
|
||||
&read_cb)) {
|
||||
done = true;
|
||||
pinnable_val->PinSelf();
|
||||
RecordTick(stats_, MEMTABLE_HIT);
|
||||
|
@ -413,8 +414,9 @@ Status DBImplSecondary::GetImpl(const ReadOptions& read_options,
|
|||
PERF_TIMER_GUARD(get_from_output_files_time);
|
||||
PinnedIteratorsManager pinned_iters_mgr;
|
||||
super_version->current->Get(
|
||||
read_options, lkey, pinnable_val, ts, &s, &merge_context,
|
||||
&max_covering_tombstone_seq, &pinned_iters_mgr, /*value_found*/ nullptr,
|
||||
read_options, lkey, pinnable_val, /*columns=*/nullptr, ts, &s,
|
||||
&merge_context, &max_covering_tombstone_seq, &pinned_iters_mgr,
|
||||
/*value_found*/ nullptr,
|
||||
/*key_exists*/ nullptr, /*seq*/ nullptr, &read_cb, /*is_blob*/ nullptr,
|
||||
/*do_merge*/ true);
|
||||
RecordTick(stats_, MEMTABLE_MISS);
|
||||
|
|
|
@ -262,9 +262,9 @@ TEST_F(DBMemTableTest, ConcurrentMergeWrite) {
|
|||
ReadOptions roptions;
|
||||
SequenceNumber max_covering_tombstone_seq = 0;
|
||||
LookupKey lkey("key", kMaxSequenceNumber);
|
||||
bool res = mem->Get(lkey, &value, /*timestamp=*/nullptr, &status,
|
||||
&merge_context, &max_covering_tombstone_seq, roptions,
|
||||
false /* immutable_memtable */);
|
||||
bool res = mem->Get(lkey, &value, /*columns=*/nullptr, /*timestamp=*/nullptr,
|
||||
&status, &merge_context, &max_covering_tombstone_seq,
|
||||
roptions, false /* immutable_memtable */);
|
||||
ASSERT_OK(status);
|
||||
ASSERT_TRUE(res);
|
||||
uint64_t ivalue = DecodeFixed64(Slice(value).data());
|
||||
|
|
|
@ -734,9 +734,9 @@ bool FlushJob::MemPurgeDecider(double threshold) {
|
|||
min_seqno_snapshot < kMaxSequenceNumber ? &min_snapshot : nullptr;
|
||||
|
||||
// Estimate if the sample entry is valid or not.
|
||||
get_res = mt->Get(lkey, &vget, nullptr, &mget_s, &merge_context,
|
||||
&max_covering_tombstone_seq, &sqno, ro,
|
||||
true /* immutable_memtable */);
|
||||
get_res = mt->Get(lkey, &vget, /*columns=*/nullptr, /*timestamp=*/nullptr,
|
||||
&mget_s, &merge_context, &max_covering_tombstone_seq,
|
||||
&sqno, ro, true /* immutable_memtable */);
|
||||
if (!get_res) {
|
||||
ROCKS_LOG_WARN(
|
||||
db_options_.info_log,
|
||||
|
@ -776,9 +776,9 @@ bool FlushJob::MemPurgeDecider(double threshold) {
|
|||
for (auto next_mem_iter = mem_iter + 1;
|
||||
next_mem_iter != std::end(mems_); next_mem_iter++) {
|
||||
if ((*next_mem_iter)
|
||||
->Get(lkey, &vget, nullptr, &mget_s, &merge_context,
|
||||
&max_covering_tombstone_seq, &sqno, ro,
|
||||
true /* immutable_memtable */)) {
|
||||
->Get(lkey, &vget, /*columns=*/nullptr, /*timestamp=*/nullptr,
|
||||
&mget_s, &merge_context, &max_covering_tombstone_seq,
|
||||
&sqno, ro, true /* immutable_memtable */)) {
|
||||
not_in_next_mems = false;
|
||||
break;
|
||||
}
|
||||
|
|
|
@ -836,6 +836,7 @@ struct Saver {
|
|||
bool* found_final_value; // Is value set correctly? Used by KeyMayExist
|
||||
bool* merge_in_progress;
|
||||
std::string* value;
|
||||
PinnableWideColumns* columns;
|
||||
SequenceNumber seq;
|
||||
std::string* timestamp;
|
||||
const MergeOperator* merge_operator;
|
||||
|
@ -866,6 +867,7 @@ static bool SaveValue(void* arg, const char* entry) {
|
|||
TEST_SYNC_POINT_CALLBACK("Memtable::SaveValue:Begin:entry", &entry);
|
||||
Saver* s = reinterpret_cast<Saver*>(arg);
|
||||
assert(s != nullptr);
|
||||
assert(!s->value || !s->columns);
|
||||
|
||||
if (s->protection_bytes_per_key > 0) {
|
||||
*(s->status) = MemTable::VerifyEntryChecksum(
|
||||
|
@ -957,7 +959,7 @@ static bool SaveValue(void* arg, const char* entry) {
|
|||
// raw merge operands to the user
|
||||
merge_context->PushOperand(
|
||||
v, s->inplace_update_support == false /* operand_pinned */);
|
||||
} else if (s->value != nullptr) {
|
||||
} else if (s->value) {
|
||||
if (type != kTypeWideColumnEntity) {
|
||||
assert(type == kTypeValue || type == kTypeBlobIndex);
|
||||
s->value->assign(v.data(), v.size());
|
||||
|
@ -969,7 +971,14 @@ static bool SaveValue(void* arg, const char* entry) {
|
|||
s->value->assign(value.data(), value.size());
|
||||
}
|
||||
}
|
||||
} else if (s->columns) {
|
||||
if (type != kTypeWideColumnEntity) {
|
||||
s->columns->SetPlainValue(v);
|
||||
} else {
|
||||
*(s->status) = s->columns->SetWideColumnValue(v);
|
||||
}
|
||||
}
|
||||
|
||||
if (s->inplace_update_support) {
|
||||
s->mem->GetLock(s->key->user_key())->ReadUnlock();
|
||||
}
|
||||
|
@ -1051,8 +1060,8 @@ static bool SaveValue(void* arg, const char* entry) {
|
|||
}
|
||||
|
||||
bool MemTable::Get(const LookupKey& key, std::string* value,
|
||||
std::string* timestamp, Status* s,
|
||||
MergeContext* merge_context,
|
||||
PinnableWideColumns* columns, std::string* timestamp,
|
||||
Status* s, MergeContext* merge_context,
|
||||
SequenceNumber* max_covering_tombstone_seq,
|
||||
SequenceNumber* seq, const ReadOptions& read_opts,
|
||||
bool immutable_memtable, ReadCallback* callback,
|
||||
|
@ -1105,8 +1114,8 @@ bool MemTable::Get(const LookupKey& key, std::string* value,
|
|||
PERF_COUNTER_ADD(bloom_memtable_hit_count, 1);
|
||||
}
|
||||
GetFromTable(key, *max_covering_tombstone_seq, do_merge, callback,
|
||||
is_blob_index, value, timestamp, s, merge_context, seq,
|
||||
&found_final_value, &merge_in_progress);
|
||||
is_blob_index, value, columns, timestamp, s, merge_context,
|
||||
seq, &found_final_value, &merge_in_progress);
|
||||
}
|
||||
|
||||
// No change to value, since we have not yet found a Put/Delete
|
||||
|
@ -1122,6 +1131,7 @@ void MemTable::GetFromTable(const LookupKey& key,
|
|||
SequenceNumber max_covering_tombstone_seq,
|
||||
bool do_merge, ReadCallback* callback,
|
||||
bool* is_blob_index, std::string* value,
|
||||
PinnableWideColumns* columns,
|
||||
std::string* timestamp, Status* s,
|
||||
MergeContext* merge_context, SequenceNumber* seq,
|
||||
bool* found_final_value, bool* merge_in_progress) {
|
||||
|
@ -1131,6 +1141,7 @@ void MemTable::GetFromTable(const LookupKey& key,
|
|||
saver.merge_in_progress = merge_in_progress;
|
||||
saver.key = &key;
|
||||
saver.value = value;
|
||||
saver.columns = columns;
|
||||
saver.timestamp = timestamp;
|
||||
saver.seq = kMaxSequenceNumber;
|
||||
saver.mem = this;
|
||||
|
@ -1207,8 +1218,9 @@ void MemTable::MultiGet(const ReadOptions& read_options, MultiGetRange* range,
|
|||
SequenceNumber dummy_seq;
|
||||
GetFromTable(*(iter->lkey), iter->max_covering_tombstone_seq, true,
|
||||
callback, &iter->is_blob_index, iter->value->GetSelf(),
|
||||
iter->timestamp, iter->s, &(iter->merge_context), &dummy_seq,
|
||||
&found_final_value, &merge_in_progress);
|
||||
/*columns=*/nullptr, iter->timestamp, iter->s,
|
||||
&(iter->merge_context), &dummy_seq, &found_final_value,
|
||||
&merge_in_progress);
|
||||
|
||||
if (!found_final_value && merge_in_progress) {
|
||||
*(iter->s) = Status::MergeInProgress();
|
||||
|
|
|
@ -259,32 +259,23 @@ class MemTable {
|
|||
// @param immutable_memtable Whether this memtable is immutable. Used
|
||||
// internally by NewRangeTombstoneIterator(). See comment above
|
||||
// NewRangeTombstoneIterator() for more detail.
|
||||
bool Get(const LookupKey& key, std::string* value, Status* s,
|
||||
bool Get(const LookupKey& key, std::string* value,
|
||||
PinnableWideColumns* columns, std::string* timestamp, Status* s,
|
||||
MergeContext* merge_context,
|
||||
SequenceNumber* max_covering_tombstone_seq, SequenceNumber* seq,
|
||||
const ReadOptions& read_opts, bool immutable_memtable,
|
||||
ReadCallback* callback = nullptr, bool* is_blob_index = nullptr,
|
||||
bool do_merge = true) {
|
||||
return Get(key, value, /*timestamp=*/nullptr, s, merge_context,
|
||||
max_covering_tombstone_seq, seq, read_opts, immutable_memtable,
|
||||
callback, is_blob_index, do_merge);
|
||||
}
|
||||
|
||||
bool Get(const LookupKey& key, std::string* value, std::string* timestamp,
|
||||
Status* s, MergeContext* merge_context,
|
||||
SequenceNumber* max_covering_tombstone_seq, SequenceNumber* seq,
|
||||
const ReadOptions& read_opts, bool immutable_memtable,
|
||||
ReadCallback* callback = nullptr, bool* is_blob_index = nullptr,
|
||||
bool do_merge = true);
|
||||
|
||||
bool Get(const LookupKey& key, std::string* value, std::string* timestamp,
|
||||
Status* s, MergeContext* merge_context,
|
||||
bool Get(const LookupKey& key, std::string* value,
|
||||
PinnableWideColumns* columns, std::string* timestamp, Status* s,
|
||||
MergeContext* merge_context,
|
||||
SequenceNumber* max_covering_tombstone_seq,
|
||||
const ReadOptions& read_opts, bool immutable_memtable,
|
||||
ReadCallback* callback = nullptr, bool* is_blob_index = nullptr,
|
||||
bool do_merge = true) {
|
||||
SequenceNumber seq;
|
||||
return Get(key, value, timestamp, s, merge_context,
|
||||
return Get(key, value, columns, timestamp, s, merge_context,
|
||||
max_covering_tombstone_seq, &seq, read_opts, immutable_memtable,
|
||||
callback, is_blob_index, do_merge);
|
||||
}
|
||||
|
@ -640,7 +631,8 @@ class MemTable {
|
|||
void GetFromTable(const LookupKey& key,
|
||||
SequenceNumber max_covering_tombstone_seq, bool do_merge,
|
||||
ReadCallback* callback, bool* is_blob_index,
|
||||
std::string* value, std::string* timestamp, Status* s,
|
||||
std::string* value, PinnableWideColumns* columns,
|
||||
std::string* timestamp, Status* s,
|
||||
MergeContext* merge_context, SequenceNumber* seq,
|
||||
bool* found_final_value, bool* merge_in_progress);
|
||||
|
||||
|
|
|
@ -105,14 +105,15 @@ int MemTableList::NumFlushed() const {
|
|||
// Return the most recent value found, if any.
|
||||
// Operands stores the list of merge operations to apply, so far.
|
||||
bool MemTableListVersion::Get(const LookupKey& key, std::string* value,
|
||||
PinnableWideColumns* columns,
|
||||
std::string* timestamp, Status* s,
|
||||
MergeContext* merge_context,
|
||||
SequenceNumber* max_covering_tombstone_seq,
|
||||
SequenceNumber* seq, const ReadOptions& read_opts,
|
||||
ReadCallback* callback, bool* is_blob_index) {
|
||||
return GetFromList(&memlist_, key, value, timestamp, s, merge_context,
|
||||
max_covering_tombstone_seq, seq, read_opts, callback,
|
||||
is_blob_index);
|
||||
return GetFromList(&memlist_, key, value, columns, timestamp, s,
|
||||
merge_context, max_covering_tombstone_seq, seq, read_opts,
|
||||
callback, is_blob_index);
|
||||
}
|
||||
|
||||
void MemTableListVersion::MultiGet(const ReadOptions& read_options,
|
||||
|
@ -131,10 +132,10 @@ bool MemTableListVersion::GetMergeOperands(
|
|||
const LookupKey& key, Status* s, MergeContext* merge_context,
|
||||
SequenceNumber* max_covering_tombstone_seq, const ReadOptions& read_opts) {
|
||||
for (MemTable* memtable : memlist_) {
|
||||
bool done =
|
||||
memtable->Get(key, /*value*/ nullptr, /*timestamp*/ nullptr, s,
|
||||
merge_context, max_covering_tombstone_seq, read_opts,
|
||||
true /* immutable_memtable */, nullptr, nullptr, false);
|
||||
bool done = memtable->Get(
|
||||
key, /*value=*/nullptr, /*columns=*/nullptr, /*timestamp=*/nullptr, s,
|
||||
merge_context, max_covering_tombstone_seq, read_opts,
|
||||
true /* immutable_memtable */, nullptr, nullptr, false);
|
||||
if (done) {
|
||||
return true;
|
||||
}
|
||||
|
@ -143,19 +144,21 @@ bool MemTableListVersion::GetMergeOperands(
|
|||
}
|
||||
|
||||
bool MemTableListVersion::GetFromHistory(
|
||||
const LookupKey& key, std::string* value, std::string* timestamp, Status* s,
|
||||
MergeContext* merge_context, SequenceNumber* max_covering_tombstone_seq,
|
||||
SequenceNumber* seq, const ReadOptions& read_opts, bool* is_blob_index) {
|
||||
return GetFromList(&memlist_history_, key, value, timestamp, s, merge_context,
|
||||
max_covering_tombstone_seq, seq, read_opts,
|
||||
const LookupKey& key, std::string* value, PinnableWideColumns* columns,
|
||||
std::string* timestamp, Status* s, MergeContext* merge_context,
|
||||
SequenceNumber* max_covering_tombstone_seq, SequenceNumber* seq,
|
||||
const ReadOptions& read_opts, bool* is_blob_index) {
|
||||
return GetFromList(&memlist_history_, key, value, columns, timestamp, s,
|
||||
merge_context, max_covering_tombstone_seq, seq, read_opts,
|
||||
nullptr /*read_callback*/, is_blob_index);
|
||||
}
|
||||
|
||||
bool MemTableListVersion::GetFromList(
|
||||
std::list<MemTable*>* list, const LookupKey& key, std::string* value,
|
||||
std::string* timestamp, Status* s, MergeContext* merge_context,
|
||||
SequenceNumber* max_covering_tombstone_seq, SequenceNumber* seq,
|
||||
const ReadOptions& read_opts, ReadCallback* callback, bool* is_blob_index) {
|
||||
PinnableWideColumns* columns, std::string* timestamp, Status* s,
|
||||
MergeContext* merge_context, SequenceNumber* max_covering_tombstone_seq,
|
||||
SequenceNumber* seq, const ReadOptions& read_opts, ReadCallback* callback,
|
||||
bool* is_blob_index) {
|
||||
*seq = kMaxSequenceNumber;
|
||||
|
||||
for (auto& memtable : *list) {
|
||||
|
@ -163,7 +166,7 @@ bool MemTableListVersion::GetFromList(
|
|||
SequenceNumber current_seq = kMaxSequenceNumber;
|
||||
|
||||
bool done =
|
||||
memtable->Get(key, value, timestamp, s, merge_context,
|
||||
memtable->Get(key, value, columns, timestamp, s, merge_context,
|
||||
max_covering_tombstone_seq, ¤t_seq, read_opts,
|
||||
true /* immutable_memtable */, callback, is_blob_index);
|
||||
if (*seq == kMaxSequenceNumber) {
|
||||
|
|
|
@ -57,19 +57,21 @@ class MemTableListVersion {
|
|||
// If any operation was found for this key, its most recent sequence number
|
||||
// will be stored in *seq on success (regardless of whether true/false is
|
||||
// returned). Otherwise, *seq will be set to kMaxSequenceNumber.
|
||||
bool Get(const LookupKey& key, std::string* value, std::string* timestamp,
|
||||
Status* s, MergeContext* merge_context,
|
||||
bool Get(const LookupKey& key, std::string* value,
|
||||
PinnableWideColumns* columns, std::string* timestamp, Status* s,
|
||||
MergeContext* merge_context,
|
||||
SequenceNumber* max_covering_tombstone_seq, SequenceNumber* seq,
|
||||
const ReadOptions& read_opts, ReadCallback* callback = nullptr,
|
||||
bool* is_blob_index = nullptr);
|
||||
|
||||
bool Get(const LookupKey& key, std::string* value, std::string* timestamp,
|
||||
Status* s, MergeContext* merge_context,
|
||||
bool Get(const LookupKey& key, std::string* value,
|
||||
PinnableWideColumns* columns, std::string* timestamp, Status* s,
|
||||
MergeContext* merge_context,
|
||||
SequenceNumber* max_covering_tombstone_seq,
|
||||
const ReadOptions& read_opts, ReadCallback* callback = nullptr,
|
||||
bool* is_blob_index = nullptr) {
|
||||
SequenceNumber seq;
|
||||
return Get(key, value, timestamp, s, merge_context,
|
||||
return Get(key, value, columns, timestamp, s, merge_context,
|
||||
max_covering_tombstone_seq, &seq, read_opts, callback,
|
||||
is_blob_index);
|
||||
}
|
||||
|
@ -89,19 +91,19 @@ class MemTableListVersion {
|
|||
// queries (such as Transaction validation) as the history may contain
|
||||
// writes that are also present in the SST files.
|
||||
bool GetFromHistory(const LookupKey& key, std::string* value,
|
||||
std::string* timestamp, Status* s,
|
||||
MergeContext* merge_context,
|
||||
PinnableWideColumns* columns, std::string* timestamp,
|
||||
Status* s, MergeContext* merge_context,
|
||||
SequenceNumber* max_covering_tombstone_seq,
|
||||
SequenceNumber* seq, const ReadOptions& read_opts,
|
||||
bool* is_blob_index = nullptr);
|
||||
bool GetFromHistory(const LookupKey& key, std::string* value,
|
||||
std::string* timestamp, Status* s,
|
||||
MergeContext* merge_context,
|
||||
PinnableWideColumns* columns, std::string* timestamp,
|
||||
Status* s, MergeContext* merge_context,
|
||||
SequenceNumber* max_covering_tombstone_seq,
|
||||
const ReadOptions& read_opts,
|
||||
bool* is_blob_index = nullptr) {
|
||||
SequenceNumber seq;
|
||||
return GetFromHistory(key, value, timestamp, s, merge_context,
|
||||
return GetFromHistory(key, value, columns, timestamp, s, merge_context,
|
||||
max_covering_tombstone_seq, &seq, read_opts,
|
||||
is_blob_index);
|
||||
}
|
||||
|
@ -158,7 +160,8 @@ class MemTableListVersion {
|
|||
bool TrimHistory(autovector<MemTable*>* to_delete, size_t usage);
|
||||
|
||||
bool GetFromList(std::list<MemTable*>* list, const LookupKey& key,
|
||||
std::string* value, std::string* timestamp, Status* s,
|
||||
std::string* value, PinnableWideColumns* columns,
|
||||
std::string* timestamp, Status* s,
|
||||
MergeContext* merge_context,
|
||||
SequenceNumber* max_covering_tombstone_seq,
|
||||
SequenceNumber* seq, const ReadOptions& read_opts,
|
||||
|
|
|
@ -238,9 +238,9 @@ TEST_F(MemTableListTest, GetTest) {
|
|||
autovector<MemTable*> to_delete;
|
||||
|
||||
LookupKey lkey("key1", seq);
|
||||
bool found = list.current()->Get(
|
||||
lkey, &value, /*timestamp*/nullptr, &s, &merge_context,
|
||||
&max_covering_tombstone_seq, ReadOptions());
|
||||
bool found = list.current()->Get(lkey, &value, /*columns=*/nullptr,
|
||||
/*timestamp=*/nullptr, &s, &merge_context,
|
||||
&max_covering_tombstone_seq, ReadOptions());
|
||||
ASSERT_FALSE(found);
|
||||
|
||||
// Create a MemTable
|
||||
|
@ -266,7 +266,7 @@ TEST_F(MemTableListTest, GetTest) {
|
|||
|
||||
// Fetch the newly written keys
|
||||
merge_context.Clear();
|
||||
found = mem->Get(LookupKey("key1", seq), &value,
|
||||
found = mem->Get(LookupKey("key1", seq), &value, /*columns*/ nullptr,
|
||||
/*timestamp*/ nullptr, &s, &merge_context,
|
||||
&max_covering_tombstone_seq, ReadOptions(),
|
||||
false /* immutable_memtable */);
|
||||
|
@ -274,7 +274,7 @@ TEST_F(MemTableListTest, GetTest) {
|
|||
ASSERT_EQ(value, "value1");
|
||||
|
||||
merge_context.Clear();
|
||||
found = mem->Get(LookupKey("key1", 2), &value,
|
||||
found = mem->Get(LookupKey("key1", 2), &value, /*columns*/ nullptr,
|
||||
/*timestamp*/ nullptr, &s, &merge_context,
|
||||
&max_covering_tombstone_seq, ReadOptions(),
|
||||
false /* immutable_memtable */);
|
||||
|
@ -282,7 +282,7 @@ TEST_F(MemTableListTest, GetTest) {
|
|||
ASSERT_TRUE(found && s.IsNotFound());
|
||||
|
||||
merge_context.Clear();
|
||||
found = mem->Get(LookupKey("key2", seq), &value,
|
||||
found = mem->Get(LookupKey("key2", seq), &value, /*columns*/ nullptr,
|
||||
/*timestamp*/ nullptr, &s, &merge_context,
|
||||
&max_covering_tombstone_seq, ReadOptions(),
|
||||
false /* immutable_memtable */);
|
||||
|
@ -319,29 +319,32 @@ TEST_F(MemTableListTest, GetTest) {
|
|||
|
||||
// Fetch keys via MemTableList
|
||||
merge_context.Clear();
|
||||
found = list.current()->Get(
|
||||
LookupKey("key1", seq), &value, /*timestamp*/nullptr, &s,
|
||||
&merge_context, &max_covering_tombstone_seq, ReadOptions());
|
||||
found =
|
||||
list.current()->Get(LookupKey("key1", seq), &value, /*columns=*/nullptr,
|
||||
/*timestamp=*/nullptr, &s, &merge_context,
|
||||
&max_covering_tombstone_seq, ReadOptions());
|
||||
ASSERT_TRUE(found && s.IsNotFound());
|
||||
|
||||
merge_context.Clear();
|
||||
found = list.current()->Get(
|
||||
LookupKey("key1", saved_seq), &value, /*timestamp*/nullptr,
|
||||
&s, &merge_context, &max_covering_tombstone_seq, ReadOptions());
|
||||
found = list.current()->Get(LookupKey("key1", saved_seq), &value,
|
||||
/*columns=*/nullptr, /*timestamp=*/nullptr, &s,
|
||||
&merge_context, &max_covering_tombstone_seq,
|
||||
ReadOptions());
|
||||
ASSERT_TRUE(s.ok() && found);
|
||||
ASSERT_EQ("value1", value);
|
||||
|
||||
merge_context.Clear();
|
||||
found = list.current()->Get(
|
||||
LookupKey("key2", seq), &value, /*timestamp*/nullptr, &s,
|
||||
&merge_context, &max_covering_tombstone_seq, ReadOptions());
|
||||
found =
|
||||
list.current()->Get(LookupKey("key2", seq), &value, /*columns=*/nullptr,
|
||||
/*timestamp=*/nullptr, &s, &merge_context,
|
||||
&max_covering_tombstone_seq, ReadOptions());
|
||||
ASSERT_TRUE(s.ok() && found);
|
||||
ASSERT_EQ(value, "value2.3");
|
||||
|
||||
merge_context.Clear();
|
||||
found = list.current()->Get(
|
||||
LookupKey("key2", 1), &value, /*timestamp*/nullptr, &s,
|
||||
&merge_context, &max_covering_tombstone_seq, ReadOptions());
|
||||
found = list.current()->Get(LookupKey("key2", 1), &value, /*columns=*/nullptr,
|
||||
/*timestamp=*/nullptr, &s, &merge_context,
|
||||
&max_covering_tombstone_seq, ReadOptions());
|
||||
ASSERT_FALSE(found);
|
||||
|
||||
ASSERT_EQ(2, list.NumNotFlushed());
|
||||
|
@ -370,9 +373,9 @@ TEST_F(MemTableListTest, GetFromHistoryTest) {
|
|||
autovector<MemTable*> to_delete;
|
||||
|
||||
LookupKey lkey("key1", seq);
|
||||
bool found = list.current()->Get(
|
||||
lkey, &value, /*timestamp*/nullptr, &s, &merge_context,
|
||||
&max_covering_tombstone_seq, ReadOptions());
|
||||
bool found = list.current()->Get(lkey, &value, /*columns=*/nullptr,
|
||||
/*timestamp=*/nullptr, &s, &merge_context,
|
||||
&max_covering_tombstone_seq, ReadOptions());
|
||||
ASSERT_FALSE(found);
|
||||
|
||||
// Create a MemTable
|
||||
|
@ -396,7 +399,7 @@ TEST_F(MemTableListTest, GetFromHistoryTest) {
|
|||
|
||||
// Fetch the newly written keys
|
||||
merge_context.Clear();
|
||||
found = mem->Get(LookupKey("key1", seq), &value,
|
||||
found = mem->Get(LookupKey("key1", seq), &value, /*columns*/ nullptr,
|
||||
/*timestamp*/ nullptr, &s, &merge_context,
|
||||
&max_covering_tombstone_seq, ReadOptions(),
|
||||
false /* immutable_memtable */);
|
||||
|
@ -404,7 +407,7 @@ TEST_F(MemTableListTest, GetFromHistoryTest) {
|
|||
ASSERT_TRUE(found && s.IsNotFound());
|
||||
|
||||
merge_context.Clear();
|
||||
found = mem->Get(LookupKey("key2", seq), &value,
|
||||
found = mem->Get(LookupKey("key2", seq), &value, /*columns*/ nullptr,
|
||||
/*timestamp*/ nullptr, &s, &merge_context,
|
||||
&max_covering_tombstone_seq, ReadOptions(),
|
||||
false /* immutable_memtable */);
|
||||
|
@ -420,15 +423,17 @@ TEST_F(MemTableListTest, GetFromHistoryTest) {
|
|||
|
||||
// Fetch keys via MemTableList
|
||||
merge_context.Clear();
|
||||
found = list.current()->Get(LookupKey("key1", seq), &value,
|
||||
/*timestamp*/nullptr, &s, &merge_context,
|
||||
&max_covering_tombstone_seq, ReadOptions());
|
||||
found =
|
||||
list.current()->Get(LookupKey("key1", seq), &value, /*columns=*/nullptr,
|
||||
/*timestamp=*/nullptr, &s, &merge_context,
|
||||
&max_covering_tombstone_seq, ReadOptions());
|
||||
ASSERT_TRUE(found && s.IsNotFound());
|
||||
|
||||
merge_context.Clear();
|
||||
found = list.current()->Get(LookupKey("key2", seq), &value,
|
||||
/*timestamp*/nullptr, &s, &merge_context,
|
||||
&max_covering_tombstone_seq, ReadOptions());
|
||||
found =
|
||||
list.current()->Get(LookupKey("key2", seq), &value, /*columns=*/nullptr,
|
||||
/*timestamp=*/nullptr, &s, &merge_context,
|
||||
&max_covering_tombstone_seq, ReadOptions());
|
||||
ASSERT_TRUE(s.ok() && found);
|
||||
ASSERT_EQ("value2.2", value);
|
||||
|
||||
|
@ -449,28 +454,32 @@ TEST_F(MemTableListTest, GetFromHistoryTest) {
|
|||
|
||||
// Verify keys are no longer in MemTableList
|
||||
merge_context.Clear();
|
||||
found = list.current()->Get(LookupKey("key1", seq), &value,
|
||||
/*timestamp*/nullptr, &s, &merge_context,
|
||||
&max_covering_tombstone_seq, ReadOptions());
|
||||
found =
|
||||
list.current()->Get(LookupKey("key1", seq), &value, /*columns=*/nullptr,
|
||||
/*timestamp=*/nullptr, &s, &merge_context,
|
||||
&max_covering_tombstone_seq, ReadOptions());
|
||||
ASSERT_FALSE(found);
|
||||
|
||||
merge_context.Clear();
|
||||
found = list.current()->Get(LookupKey("key2", seq), &value,
|
||||
/*timestamp*/nullptr, &s, &merge_context,
|
||||
&max_covering_tombstone_seq, ReadOptions());
|
||||
found =
|
||||
list.current()->Get(LookupKey("key2", seq), &value, /*columns=*/nullptr,
|
||||
/*timestamp=*/nullptr, &s, &merge_context,
|
||||
&max_covering_tombstone_seq, ReadOptions());
|
||||
ASSERT_FALSE(found);
|
||||
|
||||
// Verify keys are present in history
|
||||
merge_context.Clear();
|
||||
found = list.current()->GetFromHistory(
|
||||
LookupKey("key1", seq), &value, /*timestamp*/nullptr, &s, &merge_context,
|
||||
&max_covering_tombstone_seq, ReadOptions());
|
||||
LookupKey("key1", seq), &value, /*columns=*/nullptr,
|
||||
/*timestamp=*/nullptr, &s, &merge_context, &max_covering_tombstone_seq,
|
||||
ReadOptions());
|
||||
ASSERT_TRUE(found && s.IsNotFound());
|
||||
|
||||
merge_context.Clear();
|
||||
found = list.current()->GetFromHistory(
|
||||
LookupKey("key2", seq), &value, /*timestamp*/nullptr, &s, &merge_context,
|
||||
&max_covering_tombstone_seq, ReadOptions());
|
||||
LookupKey("key2", seq), &value, /*columns=*/nullptr,
|
||||
/*timestamp=*/nullptr, &s, &merge_context, &max_covering_tombstone_seq,
|
||||
ReadOptions());
|
||||
ASSERT_TRUE(found);
|
||||
ASSERT_EQ("value2.2", value);
|
||||
|
||||
|
@ -520,42 +529,48 @@ TEST_F(MemTableListTest, GetFromHistoryTest) {
|
|||
|
||||
// Verify keys are no longer in MemTableList
|
||||
merge_context.Clear();
|
||||
found = list.current()->Get(LookupKey("key1", seq), &value,
|
||||
/*timestamp*/nullptr, &s, &merge_context,
|
||||
&max_covering_tombstone_seq, ReadOptions());
|
||||
found =
|
||||
list.current()->Get(LookupKey("key1", seq), &value, /*columns=*/nullptr,
|
||||
/*timestamp=*/nullptr, &s, &merge_context,
|
||||
&max_covering_tombstone_seq, ReadOptions());
|
||||
ASSERT_FALSE(found);
|
||||
|
||||
merge_context.Clear();
|
||||
found = list.current()->Get(LookupKey("key2", seq), &value,
|
||||
/*timestamp*/nullptr, &s, &merge_context,
|
||||
&max_covering_tombstone_seq, ReadOptions());
|
||||
found =
|
||||
list.current()->Get(LookupKey("key2", seq), &value, /*columns=*/nullptr,
|
||||
/*timestamp=*/nullptr, &s, &merge_context,
|
||||
&max_covering_tombstone_seq, ReadOptions());
|
||||
ASSERT_FALSE(found);
|
||||
|
||||
merge_context.Clear();
|
||||
found = list.current()->Get(LookupKey("key3", seq), &value,
|
||||
/*timestamp*/nullptr, &s, &merge_context,
|
||||
&max_covering_tombstone_seq, ReadOptions());
|
||||
found =
|
||||
list.current()->Get(LookupKey("key3", seq), &value, /*columns=*/nullptr,
|
||||
/*timestamp=*/nullptr, &s, &merge_context,
|
||||
&max_covering_tombstone_seq, ReadOptions());
|
||||
ASSERT_FALSE(found);
|
||||
|
||||
// Verify that the second memtable's keys are in the history
|
||||
merge_context.Clear();
|
||||
found = list.current()->GetFromHistory(
|
||||
LookupKey("key1", seq), &value, /*timestamp*/nullptr, &s, &merge_context,
|
||||
&max_covering_tombstone_seq, ReadOptions());
|
||||
LookupKey("key1", seq), &value, /*columns=*/nullptr,
|
||||
/*timestamp=*/nullptr, &s, &merge_context, &max_covering_tombstone_seq,
|
||||
ReadOptions());
|
||||
ASSERT_TRUE(found && s.IsNotFound());
|
||||
|
||||
merge_context.Clear();
|
||||
found = list.current()->GetFromHistory(
|
||||
LookupKey("key3", seq), &value, /*timestamp*/nullptr, &s, &merge_context,
|
||||
&max_covering_tombstone_seq, ReadOptions());
|
||||
LookupKey("key3", seq), &value, /*columns=*/nullptr,
|
||||
/*timestamp=*/nullptr, &s, &merge_context, &max_covering_tombstone_seq,
|
||||
ReadOptions());
|
||||
ASSERT_TRUE(found);
|
||||
ASSERT_EQ("value3", value);
|
||||
|
||||
// Verify that key2 from the first memtable is no longer in the history
|
||||
merge_context.Clear();
|
||||
found = list.current()->Get(LookupKey("key2", seq), &value,
|
||||
/*timestamp*/nullptr, &s, &merge_context,
|
||||
&max_covering_tombstone_seq, ReadOptions());
|
||||
found =
|
||||
list.current()->Get(LookupKey("key2", seq), &value, /*columns=*/nullptr,
|
||||
/*timestamp=*/nullptr, &s, &merge_context,
|
||||
&max_covering_tombstone_seq, ReadOptions());
|
||||
ASSERT_FALSE(found);
|
||||
|
||||
// Cleanup
|
||||
|
|
|
@ -1969,7 +1969,8 @@ void Version::MultiGetBlob(
|
|||
}
|
||||
|
||||
void Version::Get(const ReadOptions& read_options, const LookupKey& k,
|
||||
PinnableSlice* value, std::string* timestamp, Status* status,
|
||||
PinnableSlice* value, PinnableWideColumns* columns,
|
||||
std::string* timestamp, Status* status,
|
||||
MergeContext* merge_context,
|
||||
SequenceNumber* max_covering_tombstone_seq,
|
||||
PinnedIteratorsManager* pinned_iters_mgr, bool* value_found,
|
||||
|
@ -2002,8 +2003,9 @@ void Version::Get(const ReadOptions& read_options, const LookupKey& k,
|
|||
GetContext get_context(
|
||||
user_comparator(), merge_operator_, info_log_, db_statistics_,
|
||||
status->ok() ? GetContext::kNotFound : GetContext::kMerge, user_key,
|
||||
do_merge ? value : nullptr, do_merge ? timestamp : nullptr, value_found,
|
||||
merge_context, do_merge, max_covering_tombstone_seq, clock_, seq,
|
||||
do_merge ? value : nullptr, do_merge ? columns : nullptr,
|
||||
do_merge ? timestamp : nullptr, value_found, merge_context, do_merge,
|
||||
max_covering_tombstone_seq, clock_, seq,
|
||||
merge_operator_ ? pinned_iters_mgr : nullptr, callback, is_blob_to_use,
|
||||
tracing_get_id, &blob_fetcher);
|
||||
|
||||
|
@ -2171,9 +2173,10 @@ void Version::MultiGet(const ReadOptions& read_options, MultiGetRange* range,
|
|||
get_ctx.emplace_back(
|
||||
user_comparator(), merge_operator_, info_log_, db_statistics_,
|
||||
iter->s->ok() ? GetContext::kNotFound : GetContext::kMerge,
|
||||
iter->ukey_with_ts, iter->value, iter->timestamp, nullptr,
|
||||
&(iter->merge_context), true, &iter->max_covering_tombstone_seq, clock_,
|
||||
nullptr, merge_operator_ ? &pinned_iters_mgr : nullptr, callback,
|
||||
iter->ukey_with_ts, iter->value, /*columns=*/nullptr, iter->timestamp,
|
||||
nullptr, &(iter->merge_context), true,
|
||||
&iter->max_covering_tombstone_seq, clock_, nullptr,
|
||||
merge_operator_ ? &pinned_iters_mgr : nullptr, callback,
|
||||
&iter->is_blob_index, tracing_mget_id, &blob_fetcher);
|
||||
// MergeInProgress status, if set, has been transferred to the get_context
|
||||
// state, so we set status to ok here. From now on, the iter status will
|
||||
|
|
|
@ -836,7 +836,8 @@ class Version {
|
|||
// REQUIRES: lock is not held
|
||||
// REQUIRES: pinned_iters_mgr != nullptr
|
||||
void Get(const ReadOptions&, const LookupKey& key, PinnableSlice* value,
|
||||
std::string* timestamp, Status* status, MergeContext* merge_context,
|
||||
PinnableWideColumns* columns, std::string* timestamp, Status* status,
|
||||
MergeContext* merge_context,
|
||||
SequenceNumber* max_covering_tombstone_seq,
|
||||
PinnedIteratorsManager* pinned_iters_mgr,
|
||||
bool* value_found = nullptr, bool* key_exists = nullptr,
|
||||
|
|
|
@ -22,10 +22,20 @@ class DBWideBasicTest : public DBTestBase {
|
|||
TEST_F(DBWideBasicTest, PutEntity) {
|
||||
Options options = GetDefaultOptions();
|
||||
|
||||
// Write a couple of wide-column entities and a plain old key-value, then read
|
||||
// them back.
|
||||
constexpr char first_key[] = "first";
|
||||
constexpr char second_key[] = "second";
|
||||
|
||||
constexpr char first_value_of_default_column[] = "hello";
|
||||
WideColumns first_columns{
|
||||
{kDefaultWideColumnName, first_value_of_default_column},
|
||||
{"attr_name1", "foo"},
|
||||
{"attr_name2", "bar"}};
|
||||
|
||||
constexpr char second_key[] = "second";
|
||||
WideColumns second_columns{{"attr_one", "two"}, {"attr_three", "four"}};
|
||||
|
||||
constexpr char third_key[] = "third";
|
||||
constexpr char third_value[] = "baz";
|
||||
|
||||
auto verify = [&]() {
|
||||
{
|
||||
|
@ -35,6 +45,13 @@ TEST_F(DBWideBasicTest, PutEntity) {
|
|||
ASSERT_EQ(result, first_value_of_default_column);
|
||||
}
|
||||
|
||||
{
|
||||
PinnableWideColumns result;
|
||||
ASSERT_OK(db_->GetEntity(ReadOptions(), db_->DefaultColumnFamily(),
|
||||
first_key, &result));
|
||||
ASSERT_EQ(result.columns(), first_columns);
|
||||
}
|
||||
|
||||
{
|
||||
PinnableSlice result;
|
||||
ASSERT_OK(db_->Get(ReadOptions(), db_->DefaultColumnFamily(), second_key,
|
||||
|
@ -43,9 +60,32 @@ TEST_F(DBWideBasicTest, PutEntity) {
|
|||
}
|
||||
|
||||
{
|
||||
constexpr size_t num_keys = 2;
|
||||
PinnableWideColumns result;
|
||||
ASSERT_OK(db_->GetEntity(ReadOptions(), db_->DefaultColumnFamily(),
|
||||
second_key, &result));
|
||||
ASSERT_EQ(result.columns(), second_columns);
|
||||
}
|
||||
|
||||
std::array<Slice, num_keys> keys{{first_key, second_key}};
|
||||
{
|
||||
PinnableSlice result;
|
||||
ASSERT_OK(db_->Get(ReadOptions(), db_->DefaultColumnFamily(), third_key,
|
||||
&result));
|
||||
ASSERT_EQ(result, third_value);
|
||||
}
|
||||
|
||||
{
|
||||
PinnableWideColumns result;
|
||||
ASSERT_OK(db_->GetEntity(ReadOptions(), db_->DefaultColumnFamily(),
|
||||
third_key, &result));
|
||||
|
||||
const WideColumns expected_columns{{kDefaultWideColumnName, third_value}};
|
||||
ASSERT_EQ(result.columns(), expected_columns);
|
||||
}
|
||||
|
||||
{
|
||||
constexpr size_t num_keys = 3;
|
||||
|
||||
std::array<Slice, num_keys> keys{{first_key, second_key, third_key}};
|
||||
std::array<PinnableSlice, num_keys> values;
|
||||
std::array<Status, num_keys> statuses;
|
||||
|
||||
|
@ -57,6 +97,9 @@ TEST_F(DBWideBasicTest, PutEntity) {
|
|||
|
||||
ASSERT_OK(statuses[1]);
|
||||
ASSERT_TRUE(values[1].empty());
|
||||
|
||||
ASSERT_OK(statuses[2]);
|
||||
ASSERT_EQ(values[2], third_value);
|
||||
}
|
||||
|
||||
{
|
||||
|
@ -74,6 +117,12 @@ TEST_F(DBWideBasicTest, PutEntity) {
|
|||
ASSERT_EQ(iter->key(), second_key);
|
||||
ASSERT_TRUE(iter->value().empty());
|
||||
|
||||
iter->Next();
|
||||
ASSERT_TRUE(iter->Valid());
|
||||
ASSERT_OK(iter->status());
|
||||
ASSERT_EQ(iter->key(), third_key);
|
||||
ASSERT_EQ(iter->value(), third_value);
|
||||
|
||||
iter->Next();
|
||||
ASSERT_FALSE(iter->Valid());
|
||||
ASSERT_OK(iter->status());
|
||||
|
@ -81,6 +130,12 @@ TEST_F(DBWideBasicTest, PutEntity) {
|
|||
iter->SeekToLast();
|
||||
ASSERT_TRUE(iter->Valid());
|
||||
ASSERT_OK(iter->status());
|
||||
ASSERT_EQ(iter->key(), third_key);
|
||||
ASSERT_EQ(iter->value(), third_value);
|
||||
|
||||
iter->Prev();
|
||||
ASSERT_TRUE(iter->Valid());
|
||||
ASSERT_OK(iter->status());
|
||||
ASSERT_EQ(iter->key(), second_key);
|
||||
ASSERT_TRUE(iter->value().empty());
|
||||
|
||||
|
@ -96,23 +151,20 @@ TEST_F(DBWideBasicTest, PutEntity) {
|
|||
}
|
||||
};
|
||||
|
||||
// Use the DB::PutEntity API
|
||||
WideColumns first_columns{
|
||||
{kDefaultWideColumnName, first_value_of_default_column},
|
||||
{"attr_name1", "foo"},
|
||||
{"attr_name2", "bar"}};
|
||||
|
||||
// Use the DB::PutEntity API to write the first entity
|
||||
ASSERT_OK(db_->PutEntity(WriteOptions(), db_->DefaultColumnFamily(),
|
||||
first_key, first_columns));
|
||||
|
||||
// Use WriteBatch
|
||||
WideColumns second_columns{{"attr_one", "two"}, {"attr_three", "four"}};
|
||||
|
||||
// Use WriteBatch to write the second entity
|
||||
WriteBatch batch;
|
||||
ASSERT_OK(
|
||||
batch.PutEntity(db_->DefaultColumnFamily(), second_key, second_columns));
|
||||
ASSERT_OK(db_->Write(WriteOptions(), &batch));
|
||||
|
||||
// Use Put to write the plain key-value
|
||||
ASSERT_OK(db_->Put(WriteOptions(), db_->DefaultColumnFamily(), third_key,
|
||||
third_value));
|
||||
|
||||
// Try reading from memtable
|
||||
verify();
|
||||
|
||||
|
|
|
@ -0,0 +1,18 @@
|
|||
// 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 "rocksdb/wide_columns.h"
|
||||
|
||||
#include "db/wide/wide_column_serialization.h"
|
||||
|
||||
namespace ROCKSDB_NAMESPACE {
|
||||
|
||||
Status PinnableWideColumns::CreateIndexForWideColumns() {
|
||||
Slice value_copy = value_;
|
||||
|
||||
return WideColumnSerialization::Deserialize(value_copy, columns_);
|
||||
}
|
||||
|
||||
} // namespace ROCKSDB_NAMESPACE
|
|
@ -567,6 +567,14 @@ class DB {
|
|||
return Get(options, DefaultColumnFamily(), key, value, timestamp);
|
||||
}
|
||||
|
||||
// UNDER CONSTRUCTION -- DO NOT USE
|
||||
virtual Status GetEntity(const ReadOptions& /* options */,
|
||||
ColumnFamilyHandle* /* column_family */,
|
||||
const Slice& /* key */,
|
||||
PinnableWideColumns* /* columns */) {
|
||||
return Status::NotSupported("GetEntity not supported");
|
||||
}
|
||||
|
||||
// Populates the `merge_operands` array with all the merge operands in the DB
|
||||
// for `key`. The `merge_operands` array will be populated in the order of
|
||||
// insertion. The number of entries populated in `merge_operands` will be
|
||||
|
|
|
@ -99,6 +99,13 @@ class StackableDB : public DB {
|
|||
return db_->Get(options, column_family, key, value);
|
||||
}
|
||||
|
||||
using DB::GetEntity;
|
||||
Status GetEntity(const ReadOptions& options,
|
||||
ColumnFamilyHandle* column_family, const Slice& key,
|
||||
PinnableWideColumns* columns) override {
|
||||
return db_->GetEntity(options, column_family, key, columns);
|
||||
}
|
||||
|
||||
using DB::GetMergeOperands;
|
||||
virtual Status GetMergeOperands(
|
||||
const ReadOptions& options, ColumnFamilyHandle* column_family,
|
||||
|
|
|
@ -11,6 +11,7 @@
|
|||
|
||||
#include "rocksdb/rocksdb_namespace.h"
|
||||
#include "rocksdb/slice.h"
|
||||
#include "rocksdb/status.h"
|
||||
|
||||
namespace ROCKSDB_NAMESPACE {
|
||||
|
||||
|
@ -69,8 +70,90 @@ inline bool operator!=(const WideColumn& lhs, const WideColumn& rhs) {
|
|||
return !(lhs == rhs);
|
||||
}
|
||||
|
||||
// A collection of wide columns.
|
||||
using WideColumns = std::vector<WideColumn>;
|
||||
|
||||
// The anonymous default wide column (an empty Slice).
|
||||
extern const Slice kDefaultWideColumnName;
|
||||
|
||||
// A self-contained collection of wide columns. Used for the results of
|
||||
// wide-column queries.
|
||||
class PinnableWideColumns {
|
||||
public:
|
||||
const WideColumns& columns() const { return columns_; }
|
||||
size_t serialized_size() const { return value_.size(); }
|
||||
|
||||
void SetPlainValue(const Slice& value);
|
||||
void SetPlainValue(const Slice& value, Cleanable* cleanable);
|
||||
|
||||
Status SetWideColumnValue(const Slice& value);
|
||||
Status SetWideColumnValue(const Slice& value, Cleanable* cleanable);
|
||||
|
||||
void Reset();
|
||||
|
||||
private:
|
||||
void CopyValue(const Slice& value);
|
||||
void PinOrCopyValue(const Slice& value, Cleanable* cleanable);
|
||||
void CreateIndexForPlainValue();
|
||||
Status CreateIndexForWideColumns();
|
||||
|
||||
PinnableSlice value_;
|
||||
WideColumns columns_;
|
||||
};
|
||||
|
||||
inline void PinnableWideColumns::CopyValue(const Slice& value) {
|
||||
value_.PinSelf(value);
|
||||
}
|
||||
|
||||
inline void PinnableWideColumns::PinOrCopyValue(const Slice& value,
|
||||
Cleanable* cleanable) {
|
||||
if (!cleanable) {
|
||||
CopyValue(value);
|
||||
return;
|
||||
}
|
||||
|
||||
value_.PinSlice(value, cleanable);
|
||||
}
|
||||
|
||||
inline void PinnableWideColumns::CreateIndexForPlainValue() {
|
||||
columns_ = WideColumns{{kDefaultWideColumnName, value_}};
|
||||
}
|
||||
|
||||
inline void PinnableWideColumns::SetPlainValue(const Slice& value) {
|
||||
CopyValue(value);
|
||||
CreateIndexForPlainValue();
|
||||
}
|
||||
|
||||
inline void PinnableWideColumns::SetPlainValue(const Slice& value,
|
||||
Cleanable* cleanable) {
|
||||
PinOrCopyValue(value, cleanable);
|
||||
CreateIndexForPlainValue();
|
||||
}
|
||||
|
||||
inline Status PinnableWideColumns::SetWideColumnValue(const Slice& value) {
|
||||
CopyValue(value);
|
||||
return CreateIndexForWideColumns();
|
||||
}
|
||||
|
||||
inline Status PinnableWideColumns::SetWideColumnValue(const Slice& value,
|
||||
Cleanable* cleanable) {
|
||||
PinOrCopyValue(value, cleanable);
|
||||
return CreateIndexForWideColumns();
|
||||
}
|
||||
|
||||
inline void PinnableWideColumns::Reset() {
|
||||
value_.Reset();
|
||||
columns_.clear();
|
||||
}
|
||||
|
||||
inline bool operator==(const PinnableWideColumns& lhs,
|
||||
const PinnableWideColumns& rhs) {
|
||||
return lhs.columns() == rhs.columns();
|
||||
}
|
||||
|
||||
inline bool operator!=(const PinnableWideColumns& lhs,
|
||||
const PinnableWideColumns& rhs) {
|
||||
return !(lhs == rhs);
|
||||
}
|
||||
|
||||
} // namespace ROCKSDB_NAMESPACE
|
||||
|
|
1
src.mk
1
src.mk
|
@ -90,6 +90,7 @@ LIB_SOURCES = \
|
|||
db/wal_edit.cc \
|
||||
db/wal_manager.cc \
|
||||
db/wide/wide_column_serialization.cc \
|
||||
db/wide/wide_columns.cc \
|
||||
db/write_batch.cc \
|
||||
db/write_batch_base.cc \
|
||||
db/write_controller.cc \
|
||||
|
|
|
@ -248,10 +248,11 @@ TEST_P(BlockBasedTableReaderTest, MultiGet) {
|
|||
autovector<KeyContext, MultiGetContext::MAX_BATCH_SIZE> key_context;
|
||||
autovector<KeyContext*, MultiGetContext::MAX_BATCH_SIZE> sorted_keys;
|
||||
for (size_t i = 0; i < keys.size(); ++i) {
|
||||
get_context.emplace_back(
|
||||
BytewiseComparator(), nullptr, nullptr, nullptr, GetContext::kNotFound,
|
||||
keys[i], &values[i], nullptr, nullptr, nullptr, true /* do_merge */,
|
||||
nullptr, nullptr, nullptr, nullptr, nullptr, nullptr);
|
||||
get_context.emplace_back(BytewiseComparator(), nullptr, nullptr, nullptr,
|
||||
GetContext::kNotFound, keys[i], &values[i],
|
||||
nullptr, nullptr, nullptr, nullptr,
|
||||
true /* do_merge */, nullptr, nullptr, nullptr,
|
||||
nullptr, nullptr, nullptr);
|
||||
key_context.emplace_back(nullptr, keys[i], &values[i], nullptr,
|
||||
&statuses.back());
|
||||
key_context.back().get_context = &get_context.back();
|
||||
|
|
|
@ -625,7 +625,7 @@ TEST(DataBlockHashIndex, BlockBoundary) {
|
|||
InternalKey seek_ikey(seek_ukey, 60, kTypeValue);
|
||||
GetContext get_context(options.comparator, nullptr, nullptr, nullptr,
|
||||
GetContext::kNotFound, seek_ukey, &value, nullptr,
|
||||
nullptr, true, nullptr, nullptr);
|
||||
nullptr, nullptr, true, nullptr, nullptr);
|
||||
|
||||
TestBoundary(ik1, v1, ik2, v2, seek_ikey, get_context, options);
|
||||
ASSERT_EQ(get_context.State(), GetContext::kFound);
|
||||
|
@ -650,7 +650,7 @@ TEST(DataBlockHashIndex, BlockBoundary) {
|
|||
InternalKey seek_ikey(seek_ukey, 60, kTypeValue);
|
||||
GetContext get_context(options.comparator, nullptr, nullptr, nullptr,
|
||||
GetContext::kNotFound, seek_ukey, &value, nullptr,
|
||||
nullptr, true, nullptr, nullptr);
|
||||
nullptr, nullptr, true, nullptr, nullptr);
|
||||
|
||||
TestBoundary(ik1, v1, ik2, v2, seek_ikey, get_context, options);
|
||||
ASSERT_EQ(get_context.State(), GetContext::kFound);
|
||||
|
@ -675,7 +675,7 @@ TEST(DataBlockHashIndex, BlockBoundary) {
|
|||
InternalKey seek_ikey(seek_ukey, 120, kTypeValue);
|
||||
GetContext get_context(options.comparator, nullptr, nullptr, nullptr,
|
||||
GetContext::kNotFound, seek_ukey, &value, nullptr,
|
||||
nullptr, true, nullptr, nullptr);
|
||||
nullptr, nullptr, true, nullptr, nullptr);
|
||||
|
||||
TestBoundary(ik1, v1, ik2, v2, seek_ikey, get_context, options);
|
||||
ASSERT_EQ(get_context.State(), GetContext::kFound);
|
||||
|
@ -700,7 +700,7 @@ TEST(DataBlockHashIndex, BlockBoundary) {
|
|||
InternalKey seek_ikey(seek_ukey, 5, kTypeValue);
|
||||
GetContext get_context(options.comparator, nullptr, nullptr, nullptr,
|
||||
GetContext::kNotFound, seek_ukey, &value, nullptr,
|
||||
nullptr, true, nullptr, nullptr);
|
||||
nullptr, nullptr, true, nullptr, nullptr);
|
||||
|
||||
TestBoundary(ik1, v1, ik2, v2, seek_ikey, get_context, options);
|
||||
ASSERT_EQ(get_context.State(), GetContext::kNotFound);
|
||||
|
|
|
@ -119,7 +119,8 @@ class CuckooReaderTest : public testing::Test {
|
|||
PinnableSlice value;
|
||||
GetContext get_context(ucomp, nullptr, nullptr, nullptr,
|
||||
GetContext::kNotFound, Slice(user_keys[i]), &value,
|
||||
nullptr, nullptr, true, nullptr, nullptr);
|
||||
nullptr, nullptr, nullptr, nullptr, true, nullptr,
|
||||
nullptr);
|
||||
ASSERT_OK(
|
||||
reader.Get(ReadOptions(), Slice(keys[i]), &get_context, nullptr));
|
||||
ASSERT_STREQ(values[i].c_str(), value.data());
|
||||
|
@ -341,8 +342,8 @@ TEST_F(CuckooReaderTest, WhenKeyNotFound) {
|
|||
AppendInternalKey(¬_found_key, ikey);
|
||||
PinnableSlice value;
|
||||
GetContext get_context(ucmp, nullptr, nullptr, nullptr, GetContext::kNotFound,
|
||||
Slice(not_found_key), &value, nullptr, nullptr, true,
|
||||
nullptr, nullptr);
|
||||
Slice(not_found_key), &value, nullptr, nullptr,
|
||||
nullptr, nullptr, true, nullptr, nullptr);
|
||||
ASSERT_OK(
|
||||
reader.Get(ReadOptions(), Slice(not_found_key), &get_context, nullptr));
|
||||
ASSERT_TRUE(value.empty());
|
||||
|
@ -356,7 +357,8 @@ TEST_F(CuckooReaderTest, WhenKeyNotFound) {
|
|||
value.Reset();
|
||||
GetContext get_context2(ucmp, nullptr, nullptr, nullptr,
|
||||
GetContext::kNotFound, Slice(not_found_key2), &value,
|
||||
nullptr, nullptr, true, nullptr, nullptr);
|
||||
nullptr, nullptr, nullptr, nullptr, true, nullptr,
|
||||
nullptr);
|
||||
ASSERT_OK(
|
||||
reader.Get(ReadOptions(), Slice(not_found_key2), &get_context2, nullptr));
|
||||
ASSERT_TRUE(value.empty());
|
||||
|
@ -370,9 +372,9 @@ TEST_F(CuckooReaderTest, WhenKeyNotFound) {
|
|||
AddHashLookups(ExtractUserKey(unused_key).ToString(),
|
||||
kNumHashFunc, kNumHashFunc);
|
||||
value.Reset();
|
||||
GetContext get_context3(ucmp, nullptr, nullptr, nullptr,
|
||||
GetContext::kNotFound, Slice(unused_key), &value,
|
||||
nullptr, nullptr, true, nullptr, nullptr);
|
||||
GetContext get_context3(
|
||||
ucmp, nullptr, nullptr, nullptr, GetContext::kNotFound, Slice(unused_key),
|
||||
&value, nullptr, nullptr, nullptr, nullptr, true, nullptr, nullptr);
|
||||
ASSERT_OK(
|
||||
reader.Get(ReadOptions(), Slice(unused_key), &get_context3, nullptr));
|
||||
ASSERT_TRUE(value.empty());
|
||||
|
@ -447,7 +449,7 @@ void WriteFile(const std::vector<std::string>& keys,
|
|||
// Assume only the fast path is triggered
|
||||
GetContext get_context(nullptr, nullptr, nullptr, nullptr,
|
||||
GetContext::kNotFound, Slice(), &value, nullptr,
|
||||
nullptr, true, nullptr, nullptr);
|
||||
nullptr, nullptr, true, nullptr, nullptr);
|
||||
for (uint64_t i = 0; i < num; ++i) {
|
||||
value.Reset();
|
||||
value.clear();
|
||||
|
@ -496,7 +498,7 @@ void ReadKeys(uint64_t num, uint32_t batch_size) {
|
|||
// Assume only the fast path is triggered
|
||||
GetContext get_context(nullptr, nullptr, nullptr, nullptr,
|
||||
GetContext::kNotFound, Slice(), &value, nullptr,
|
||||
nullptr, true, nullptr, nullptr);
|
||||
nullptr, nullptr, true, nullptr, nullptr);
|
||||
uint64_t start_time = env->NowMicros();
|
||||
if (batch_size > 0) {
|
||||
for (uint64_t i = 0; i < num; i += batch_size) {
|
||||
|
|
|
@ -41,17 +41,15 @@ void appendToReplayLog(std::string* replay_log, ValueType type, Slice value) {
|
|||
|
||||
} // namespace
|
||||
|
||||
GetContext::GetContext(const Comparator* ucmp,
|
||||
const MergeOperator* merge_operator, Logger* logger,
|
||||
Statistics* statistics, GetState init_state,
|
||||
const Slice& user_key, PinnableSlice* pinnable_val,
|
||||
std::string* timestamp, bool* value_found,
|
||||
MergeContext* merge_context, bool do_merge,
|
||||
SequenceNumber* _max_covering_tombstone_seq,
|
||||
SystemClock* clock, SequenceNumber* seq,
|
||||
PinnedIteratorsManager* _pinned_iters_mgr,
|
||||
ReadCallback* callback, bool* is_blob_index,
|
||||
uint64_t tracing_get_id, BlobFetcher* blob_fetcher)
|
||||
GetContext::GetContext(
|
||||
const Comparator* ucmp, const MergeOperator* merge_operator, Logger* logger,
|
||||
Statistics* statistics, GetState init_state, const Slice& user_key,
|
||||
PinnableSlice* pinnable_val, PinnableWideColumns* columns,
|
||||
std::string* timestamp, bool* value_found, MergeContext* merge_context,
|
||||
bool do_merge, SequenceNumber* _max_covering_tombstone_seq,
|
||||
SystemClock* clock, SequenceNumber* seq,
|
||||
PinnedIteratorsManager* _pinned_iters_mgr, ReadCallback* callback,
|
||||
bool* is_blob_index, uint64_t tracing_get_id, BlobFetcher* blob_fetcher)
|
||||
: ucmp_(ucmp),
|
||||
merge_operator_(merge_operator),
|
||||
logger_(logger),
|
||||
|
@ -59,6 +57,7 @@ GetContext::GetContext(const Comparator* ucmp,
|
|||
state_(init_state),
|
||||
user_key_(user_key),
|
||||
pinnable_val_(pinnable_val),
|
||||
columns_(columns),
|
||||
timestamp_(timestamp),
|
||||
value_found_(value_found),
|
||||
merge_context_(merge_context),
|
||||
|
@ -78,18 +77,22 @@ GetContext::GetContext(const Comparator* ucmp,
|
|||
sample_ = should_sample_file_read();
|
||||
}
|
||||
|
||||
GetContext::GetContext(
|
||||
const Comparator* ucmp, const MergeOperator* merge_operator, Logger* logger,
|
||||
Statistics* statistics, GetState init_state, const Slice& user_key,
|
||||
PinnableSlice* pinnable_val, bool* value_found, MergeContext* merge_context,
|
||||
bool do_merge, SequenceNumber* _max_covering_tombstone_seq,
|
||||
SystemClock* clock, SequenceNumber* seq,
|
||||
PinnedIteratorsManager* _pinned_iters_mgr, ReadCallback* callback,
|
||||
bool* is_blob_index, uint64_t tracing_get_id, BlobFetcher* blob_fetcher)
|
||||
GetContext::GetContext(const Comparator* ucmp,
|
||||
const MergeOperator* merge_operator, Logger* logger,
|
||||
Statistics* statistics, GetState init_state,
|
||||
const Slice& user_key, PinnableSlice* pinnable_val,
|
||||
PinnableWideColumns* columns, bool* value_found,
|
||||
MergeContext* merge_context, bool do_merge,
|
||||
SequenceNumber* _max_covering_tombstone_seq,
|
||||
SystemClock* clock, SequenceNumber* seq,
|
||||
PinnedIteratorsManager* _pinned_iters_mgr,
|
||||
ReadCallback* callback, bool* is_blob_index,
|
||||
uint64_t tracing_get_id, BlobFetcher* blob_fetcher)
|
||||
: GetContext(ucmp, merge_operator, logger, statistics, init_state, user_key,
|
||||
pinnable_val, nullptr, value_found, merge_context, do_merge,
|
||||
_max_covering_tombstone_seq, clock, seq, _pinned_iters_mgr,
|
||||
callback, is_blob_index, tracing_get_id, blob_fetcher) {}
|
||||
pinnable_val, columns, /*timestamp=*/nullptr, value_found,
|
||||
merge_context, do_merge, _max_covering_tombstone_seq, clock,
|
||||
seq, _pinned_iters_mgr, callback, is_blob_index,
|
||||
tracing_get_id, blob_fetcher) {}
|
||||
|
||||
// Called from TableCache::Get and Table::Get when file/block in which
|
||||
// key may exist are not there in TableCache/BlockCache respectively. In this
|
||||
|
@ -291,6 +294,15 @@ bool GetContext::SaveValue(const ParsedInternalKey& parsed_key,
|
|||
// Otherwise copy the value
|
||||
pinnable_val_->PinSelf(value_to_use);
|
||||
}
|
||||
} else if (columns_ != nullptr) {
|
||||
if (type == kTypeWideColumnEntity) {
|
||||
if (!columns_->SetWideColumnValue(value, value_pinner).ok()) {
|
||||
state_ = kCorrupt;
|
||||
return false;
|
||||
}
|
||||
} else {
|
||||
columns_->SetPlainValue(value, value_pinner);
|
||||
}
|
||||
}
|
||||
} else {
|
||||
// It means this function is called as part of DB GetMergeOperands
|
||||
|
|
|
@ -15,6 +15,7 @@ class Comparator;
|
|||
class Logger;
|
||||
class MergeContext;
|
||||
class MergeOperator;
|
||||
class PinnableWideColumns;
|
||||
class PinnedIteratorsManager;
|
||||
class Statistics;
|
||||
class SystemClock;
|
||||
|
@ -101,7 +102,8 @@ class GetContext {
|
|||
// merge_context and they are never merged. The value pointer is untouched.
|
||||
GetContext(const Comparator* ucmp, const MergeOperator* merge_operator,
|
||||
Logger* logger, Statistics* statistics, GetState init_state,
|
||||
const Slice& user_key, PinnableSlice* value, bool* value_found,
|
||||
const Slice& user_key, PinnableSlice* value,
|
||||
PinnableWideColumns* columns, bool* value_found,
|
||||
MergeContext* merge_context, bool do_merge,
|
||||
SequenceNumber* max_covering_tombstone_seq, SystemClock* clock,
|
||||
SequenceNumber* seq = nullptr,
|
||||
|
@ -111,8 +113,8 @@ class GetContext {
|
|||
GetContext(const Comparator* ucmp, const MergeOperator* merge_operator,
|
||||
Logger* logger, Statistics* statistics, GetState init_state,
|
||||
const Slice& user_key, PinnableSlice* value,
|
||||
std::string* timestamp, bool* value_found,
|
||||
MergeContext* merge_context, bool do_merge,
|
||||
PinnableWideColumns* columns, std::string* timestamp,
|
||||
bool* value_found, MergeContext* merge_context, bool do_merge,
|
||||
SequenceNumber* max_covering_tombstone_seq, SystemClock* clock,
|
||||
SequenceNumber* seq = nullptr,
|
||||
PinnedIteratorsManager* _pinned_iters_mgr = nullptr,
|
||||
|
@ -186,6 +188,7 @@ class GetContext {
|
|||
GetState state_;
|
||||
Slice user_key_;
|
||||
PinnableSlice* pinnable_val_;
|
||||
PinnableWideColumns* columns_;
|
||||
std::string* timestamp_;
|
||||
bool* value_found_; // Is value set correctly? Used by KeyMayExist
|
||||
MergeContext* merge_context_;
|
||||
|
|
|
@ -177,8 +177,8 @@ void TableReaderBenchmark(Options& opts, EnvOptions& env_options,
|
|||
GetContext get_context(
|
||||
ioptions.user_comparator, ioptions.merge_operator.get(),
|
||||
ioptions.logger, ioptions.stats, GetContext::kNotFound,
|
||||
Slice(key), &value, nullptr, &merge_context, true,
|
||||
&max_covering_tombstone_seq, clock);
|
||||
Slice(key), &value, /*columns=*/nullptr, /*timestamp=*/nullptr,
|
||||
&merge_context, true, &max_covering_tombstone_seq, clock);
|
||||
s = table_reader->Get(read_options, key, &get_context, nullptr);
|
||||
} else {
|
||||
s = db->Get(read_options, key, &result);
|
||||
|
|
|
@ -3036,8 +3036,8 @@ TEST_P(BlockBasedTableTest, TracingGetTest) {
|
|||
PinnableSlice value;
|
||||
GetContext get_context(options.comparator, nullptr, nullptr, nullptr,
|
||||
GetContext::kNotFound, user_key, &value, nullptr,
|
||||
nullptr, true, nullptr, nullptr, nullptr, nullptr,
|
||||
nullptr, nullptr, /*tracing_get_id=*/i);
|
||||
nullptr, nullptr, true, nullptr, nullptr, nullptr,
|
||||
nullptr, nullptr, nullptr, /*tracing_get_id=*/i);
|
||||
get_perf_context()->Reset();
|
||||
ASSERT_OK(c.GetTableReader()->Get(ReadOptions(), encoded_key, &get_context,
|
||||
moptions.prefix_extractor.get()));
|
||||
|
@ -3293,7 +3293,7 @@ TEST_P(BlockBasedTableTest, BlockCacheDisabledTest) {
|
|||
{
|
||||
GetContext get_context(options.comparator, nullptr, nullptr, nullptr,
|
||||
GetContext::kNotFound, Slice(), nullptr, nullptr,
|
||||
nullptr, true, nullptr, nullptr);
|
||||
nullptr, nullptr, true, nullptr, nullptr);
|
||||
// a hack that just to trigger BlockBasedTable::GetFilter.
|
||||
ASSERT_OK(reader->Get(ReadOptions(), "non-exist-key", &get_context,
|
||||
moptions.prefix_extractor.get()));
|
||||
|
@ -3471,7 +3471,7 @@ TEST_P(BlockBasedTableTest, FilterBlockInBlockCache) {
|
|||
PinnableSlice value;
|
||||
GetContext get_context(options.comparator, nullptr, nullptr, nullptr,
|
||||
GetContext::kNotFound, user_key, &value, nullptr,
|
||||
nullptr, true, nullptr, nullptr);
|
||||
nullptr, nullptr, true, nullptr, nullptr);
|
||||
ASSERT_OK(reader->Get(ReadOptions(), internal_key.Encode(), &get_context,
|
||||
moptions4.prefix_extractor.get()));
|
||||
ASSERT_STREQ(value.data(), "hello");
|
||||
|
@ -3558,7 +3558,7 @@ TEST_P(BlockBasedTableTest, BlockReadCountTest) {
|
|||
{
|
||||
GetContext get_context(options.comparator, nullptr, nullptr, nullptr,
|
||||
GetContext::kNotFound, user_key, &value, nullptr,
|
||||
nullptr, true, nullptr, nullptr);
|
||||
nullptr, nullptr, true, nullptr, nullptr);
|
||||
get_perf_context()->Reset();
|
||||
ASSERT_OK(reader->Get(ReadOptions(), encoded_key, &get_context,
|
||||
moptions.prefix_extractor.get()));
|
||||
|
@ -3584,7 +3584,7 @@ TEST_P(BlockBasedTableTest, BlockReadCountTest) {
|
|||
{
|
||||
GetContext get_context(options.comparator, nullptr, nullptr, nullptr,
|
||||
GetContext::kNotFound, user_key, &value, nullptr,
|
||||
nullptr, true, nullptr, nullptr);
|
||||
nullptr, nullptr, true, nullptr, nullptr);
|
||||
get_perf_context()->Reset();
|
||||
ASSERT_OK(reader->Get(ReadOptions(), encoded_key, &get_context,
|
||||
moptions.prefix_extractor.get()));
|
||||
|
@ -5149,7 +5149,7 @@ TEST_P(BlockBasedTableTest, DataBlockHashIndex) {
|
|||
std::string user_key = ExtractUserKey(kv.first).ToString();
|
||||
GetContext get_context(options.comparator, nullptr, nullptr, nullptr,
|
||||
GetContext::kNotFound, user_key, &value, nullptr,
|
||||
nullptr, true, nullptr, nullptr);
|
||||
nullptr, nullptr, true, nullptr, nullptr);
|
||||
ASSERT_OK(reader->Get(ro, kv.first, &get_context,
|
||||
moptions.prefix_extractor.get()));
|
||||
ASSERT_EQ(get_context.State(), GetContext::kFound);
|
||||
|
@ -5175,7 +5175,7 @@ TEST_P(BlockBasedTableTest, DataBlockHashIndex) {
|
|||
PinnableSlice value;
|
||||
GetContext get_context(options.comparator, nullptr, nullptr, nullptr,
|
||||
GetContext::kNotFound, user_key, &value, nullptr,
|
||||
nullptr, true, nullptr, nullptr);
|
||||
nullptr, nullptr, true, nullptr, nullptr);
|
||||
ASSERT_OK(reader->Get(ro, encoded_key, &get_context,
|
||||
moptions.prefix_extractor.get()));
|
||||
ASSERT_EQ(get_context.State(), GetContext::kNotFound);
|
||||
|
|
Loading…
Reference in New Issue