jabberwocky238/rocksdb
1
0
Fork 0
mirror of https://github.com/facebook/rocksdb.git synced 2026-07-07 14:47:40 +08:00
Code Issues Packages Projects Releases Wiki Activity
Files
0f15da1c2db1d5583d0c7200208b6465357f95fc
rocksdb/table/get_context.cc
T
Xingbo Wang 4707775ae9 Fix GetContext status propagation and blob-backed wide-column merge operands (#14640) 
Summary:
- propagate lower-level read and merge failures through `GetContext` via `read_status`, so `Get` and `GetEntity` preserve the original error instead of synthesizing `Corruption` when blob-backed reads or merge resolution fail
- teach `GetMergeOperands` to resolve blob-backed default columns from wide-column entities, covering both the direct base-value path and the merge-plus-base path
- add regression coverage for blob-read IO errors during `Get`/`GetEntity` merge resolution and for `GetMergeOperands` on blob-backed wide-column entities
- fix the `DBFlushTest.MemPurgeCorrectLogNumberAndSSTFileCreation` test race by waiting for flush callbacks and cleaning up sync points

## Testing

- `make db_blob_basic_test -j14`
- `/usr/bin/perl -e 'alarm shift; exec ARGV' 60 ./db_blob_basic_test --gtest_filter='DBBlobBasicTest/DBBlobBasicIOErrorTest.GetBlob_IOError/*:DBBlobBasicTest/DBBlobBasicIOErrorTest.GetEntityMergeWithBlobBaseIOError/*'`

## Task
T265824017, T265415808

Pull Request resolved: https://github.com/facebook/rocksdb/pull/14640

Reviewed By: anand1976

Differential Revision: D101690700

Pulled By: xingbowang

fbshipit-source-id: 2b6fc357b37a01efa72a2d54dcff55be8992f42a
2026-05-12 15:29:27 -07:00

835 lines
31 KiB
C++
Raw Blame History

// 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).
#include "table/get_context.h"
#include <vector>
#include "db/blob/blob_fetcher.h"
#include "db/blob/blob_index.h"
#include "db/merge_helper.h"
#include "db/pinned_iterators_manager.h"
#include "db/read_callback.h"
#include "db/wide/wide_column_serialization.h"
#include "db/wide/wide_columns_helper.h"
#include "monitoring/file_read_sample.h"
#include "monitoring/perf_context_imp.h"
#include "monitoring/statistics_impl.h"
#include "port/likely.h"
#include "rocksdb/merge_operator.h"
#include "rocksdb/statistics.h"
#include "rocksdb/status.h"
#include "rocksdb/system_clock.h"
namespace ROCKSDB_NAMESPACE {
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),
statistics_(statistics),
state_(init_state),
user_key_(user_key),
pinnable_val_(pinnable_val),
columns_(columns),
timestamp_(timestamp),
value_found_(value_found),
merge_context_(merge_context),
max_covering_tombstone_seq_(_max_covering_tombstone_seq),
clock_(clock),
seq_(seq),
replay_log_(nullptr),
pinned_iters_mgr_(_pinned_iters_mgr),
callback_(callback),
do_merge_(do_merge),
is_blob_index_(is_blob_index),
tracing_get_id_(tracing_get_id),
blob_fetcher_(blob_fetcher) {
if (seq_) {
*seq_ = kMaxSequenceNumber;
}
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,
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, 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) {}
void GetContext::appendToReplayLog(ValueType type, Slice value, Slice ts) {
if (replay_log_) {
if (replay_log_->empty()) {
// Optimization: in the common case of only one operation in the
// log, we allocate the exact amount of space needed.
replay_log_->reserve(1 + VarintLength(value.size()) + value.size());
}
replay_log_->push_back(type);
PutLengthPrefixedSlice(replay_log_, value);
// If cf enables ts, there should always be a ts following each value
if (ucmp_->timestamp_size() > 0) {
assert(ts.size() == ucmp_->timestamp_size());
PutLengthPrefixedSlice(replay_log_, ts);
}
}
}
// Called from TableCache::Get and Table::Get when file/block in which
// key may exist are not there in TableCache/BlockCache respectively. In this
// case we can't guarantee that key does not exist and are not permitted to do
// IO to be certain.Set the status=kFound and value_found=false to let the
// caller know that key may exist but is not there in memory
void GetContext::MarkKeyMayExist() {
state_ = kFound;
if (value_found_ != nullptr) {
*value_found_ = false;
}
}
Status GetContext::SaveWideColumnEntityToPinnable(const Slice& user_key,
const Slice& entity,
Cleanable* value_pinner) {
assert(pinnable_val_ != nullptr);
// Try the fast path first: GetValueOfDefaultColumn handles both V1 and V2
// entities with inline default column without full deserialization. It
// returns NotSupported only when the default column is a blob reference.
Slice value_of_default;
Slice entity_ref = entity;
Status status = WideColumnSerialization::GetValueOfDefaultColumn(
entity_ref, value_of_default);
if (status.ok()) {
if (LIKELY(value_pinner != nullptr)) {
pinnable_val_->PinSlice(value_of_default, value_pinner);
} else {
pinnable_val_->PinSelf(value_of_default);
}
} else if (status.IsNotSupported()) {
if (blob_fetcher_ == nullptr) {
return Status::Corruption(
"Cannot resolve blob-backed default column without a blob fetcher");
}
// Default column is a blob reference, so resolve it into the output value.
bool resolved = false;
status = WideColumnSerialization::GetValueOfDefaultColumnResolvingBlobs(
entity, user_key, blob_fetcher_, *pinnable_val_, resolved);
}
return status;
}
Status GetContext::SaveWideColumnEntityToColumns(const Slice& user_key,
const Slice& entity,
Cleanable* value_pinner) {
assert(columns_ != nullptr);
bool has_blob_columns = false;
Status status =
WideColumnSerialization::HasBlobColumns(entity, has_blob_columns);
if (!status.ok()) {
return status;
}
if (!has_blob_columns) {
return columns_->SetWideColumnValue(entity, value_pinner);
}
std::vector<WideColumn> entity_columns;
std::vector<std::pair<size_t, BlobIndex>> blob_cols;
Slice entity_ref = entity;
status = WideColumnSerialization::DeserializeV2(entity_ref, entity_columns,
blob_cols);
if (status.ok()) {
// HasBlobColumns() and DeserializeV2() must agree on whether this entity
// contains blob-valued columns.
assert(!blob_cols.empty());
if (blob_fetcher_ == nullptr) {
return Status::Corruption(
"Cannot resolve blob-backed wide-column entity without a blob "
"fetcher");
}
// TODO: Add lazy resolution support for GetEntity point lookups. This
// requires SuperVersion pinning on PinnableWideColumns to keep the
// Version* alive after GetImpl returns. Currently, lazy_column_resolution
// only takes effect for iterators (DBIter path).
//
// Eager path: resolve blob columns inline to avoid intermediate
// std::string copies per blob value. Keep fetched blob values as
// PinnableSlice.
std::vector<PinnableSlice> resolved_blob_values(blob_cols.size());
for (size_t bi = 0; bi < blob_cols.size() && status.ok(); ++bi) {
const BlobIndex& blob_idx = blob_cols[bi].second;
if (blob_idx.IsInlined()) {
resolved_blob_values[bi].PinSelf(blob_idx.value());
continue;
}
status = blob_fetcher_->FetchBlob(
user_key, blob_idx, nullptr /* prefetch_buffer */,
&resolved_blob_values[bi], nullptr /* bytes_read */);
}
if (status.ok()) {
WideColumns result_columns;
result_columns.reserve(entity_columns.size());
size_t blob_cursor = 0;
for (size_t ci = 0; ci < entity_columns.size(); ++ci) {
if (blob_cursor < blob_cols.size() &&
blob_cols[blob_cursor].first == ci) {
result_columns.emplace_back(entity_columns[ci].name(),
Slice(resolved_blob_values[blob_cursor]));
++blob_cursor;
} else {
result_columns.emplace_back(entity_columns[ci].name(),
entity_columns[ci].value());
}
}
std::string resolved_entity;
status =
WideColumnSerialization::Serialize(result_columns, resolved_entity);
if (status.ok()) {
// TODO: A combined SerializeAndBuildIndex method could avoid the
// serialize + deserialize round trip inside
// SetWideColumnValue -> CreateIndexForWideColumns.
return columns_->SetWideColumnValue(std::move(resolved_entity));
}
}
}
return status;
}
Status GetContext::PushWideColumnEntityDefaultOperand(const Slice& user_key,
const Slice& entity,
Cleanable* value_pinner) {
Slice value_of_default;
Slice entity_ref = entity;
Status status = WideColumnSerialization::GetValueOfDefaultColumn(
entity_ref, value_of_default);
if (status.ok()) {
push_operand(value_of_default, value_pinner);
return status;
}
if (!status.IsNotSupported()) {
return status;
}
if (blob_fetcher_ == nullptr) {
return Status::Corruption(
"Cannot resolve blob-backed default column without a blob fetcher");
}
PinnableSlice resolved_default;
bool resolved = false;
status = WideColumnSerialization::GetValueOfDefaultColumnResolvingBlobs(
entity, user_key, blob_fetcher_, resolved_default, resolved);
if (status.ok()) {
// Resolved blob values are backed by this stack-local PinnableSlice, so
// copy them into MergeContext instead of pinning their storage.
push_operand(Slice(resolved_default), nullptr);
}
return status;
}
void GetContext::SaveValue(const Slice& value, SequenceNumber /*seq*/,
Cleanable* value_pinner) {
assert(state_ == kNotFound);
assert(ucmp_->timestamp_size() == 0);
TEST_SYNC_POINT_CALLBACK("GetContext::SaveValue::Simple", this);
appendToReplayLog(kTypeValue, value, Slice());
state_ = kFound;
if (LIKELY(pinnable_val_ != nullptr)) {
if (LIKELY(value_pinner != nullptr)) {
pinnable_val_->PinSlice(value, value_pinner);
} else {
pinnable_val_->PinSelf(value);
}
}
}
void GetContext::ReportCounters() {
if (get_context_stats_.num_cache_hit > 0) {
RecordTick(statistics_, BLOCK_CACHE_HIT, get_context_stats_.num_cache_hit);
}
if (get_context_stats_.num_cache_index_hit > 0) {
RecordTick(statistics_, BLOCK_CACHE_INDEX_HIT,
get_context_stats_.num_cache_index_hit);
}
if (get_context_stats_.num_cache_data_hit > 0) {
RecordTick(statistics_, BLOCK_CACHE_DATA_HIT,
get_context_stats_.num_cache_data_hit);
}
if (get_context_stats_.num_cache_filter_hit > 0) {
RecordTick(statistics_, BLOCK_CACHE_FILTER_HIT,
get_context_stats_.num_cache_filter_hit);
}
if (get_context_stats_.num_cache_compression_dict_hit > 0) {
RecordTick(statistics_, BLOCK_CACHE_COMPRESSION_DICT_HIT,
get_context_stats_.num_cache_compression_dict_hit);
}
if (get_context_stats_.num_cache_index_miss > 0) {
RecordTick(statistics_, BLOCK_CACHE_INDEX_MISS,
get_context_stats_.num_cache_index_miss);
}
if (get_context_stats_.num_cache_filter_miss > 0) {
RecordTick(statistics_, BLOCK_CACHE_FILTER_MISS,
get_context_stats_.num_cache_filter_miss);
}
if (get_context_stats_.num_cache_data_miss > 0) {
RecordTick(statistics_, BLOCK_CACHE_DATA_MISS,
get_context_stats_.num_cache_data_miss);
}
if (get_context_stats_.num_cache_compression_dict_miss > 0) {
RecordTick(statistics_, BLOCK_CACHE_COMPRESSION_DICT_MISS,
get_context_stats_.num_cache_compression_dict_miss);
}
if (get_context_stats_.num_cache_bytes_read > 0) {
RecordTick(statistics_, BLOCK_CACHE_BYTES_READ,
get_context_stats_.num_cache_bytes_read);
}
if (get_context_stats_.num_cache_miss > 0) {
RecordTick(statistics_, BLOCK_CACHE_MISS,
get_context_stats_.num_cache_miss);
}
if (get_context_stats_.num_cache_add > 0) {
RecordTick(statistics_, BLOCK_CACHE_ADD, get_context_stats_.num_cache_add);
}
if (get_context_stats_.num_cache_add_redundant > 0) {
RecordTick(statistics_, BLOCK_CACHE_ADD_REDUNDANT,
get_context_stats_.num_cache_add_redundant);
}
if (get_context_stats_.num_cache_bytes_write > 0) {
RecordTick(statistics_, BLOCK_CACHE_BYTES_WRITE,
get_context_stats_.num_cache_bytes_write);
}
if (get_context_stats_.num_cache_index_add > 0) {
RecordTick(statistics_, BLOCK_CACHE_INDEX_ADD,
get_context_stats_.num_cache_index_add);
}
if (get_context_stats_.num_cache_index_add_redundant > 0) {
RecordTick(statistics_, BLOCK_CACHE_INDEX_ADD_REDUNDANT,
get_context_stats_.num_cache_index_add_redundant);
}
if (get_context_stats_.num_cache_index_bytes_insert > 0) {
RecordTick(statistics_, BLOCK_CACHE_INDEX_BYTES_INSERT,
get_context_stats_.num_cache_index_bytes_insert);
}
if (get_context_stats_.num_cache_data_add > 0) {
RecordTick(statistics_, BLOCK_CACHE_DATA_ADD,
get_context_stats_.num_cache_data_add);
}
if (get_context_stats_.num_cache_data_add_redundant > 0) {
RecordTick(statistics_, BLOCK_CACHE_DATA_ADD_REDUNDANT,
get_context_stats_.num_cache_data_add_redundant);
}
if (get_context_stats_.num_cache_data_bytes_insert > 0) {
RecordTick(statistics_, BLOCK_CACHE_DATA_BYTES_INSERT,
get_context_stats_.num_cache_data_bytes_insert);
}
if (get_context_stats_.num_cache_filter_add > 0) {
RecordTick(statistics_, BLOCK_CACHE_FILTER_ADD,
get_context_stats_.num_cache_filter_add);
}
if (get_context_stats_.num_cache_filter_add_redundant > 0) {
RecordTick(statistics_, BLOCK_CACHE_FILTER_ADD_REDUNDANT,
get_context_stats_.num_cache_filter_add_redundant);
}
if (get_context_stats_.num_cache_filter_bytes_insert > 0) {
RecordTick(statistics_, BLOCK_CACHE_FILTER_BYTES_INSERT,
get_context_stats_.num_cache_filter_bytes_insert);
}
if (get_context_stats_.num_cache_compression_dict_add > 0) {
RecordTick(statistics_, BLOCK_CACHE_COMPRESSION_DICT_ADD,
get_context_stats_.num_cache_compression_dict_add);
}
if (get_context_stats_.num_cache_compression_dict_add_redundant > 0) {
RecordTick(statistics_, BLOCK_CACHE_COMPRESSION_DICT_ADD_REDUNDANT,
get_context_stats_.num_cache_compression_dict_add_redundant);
}
if (get_context_stats_.num_cache_compression_dict_bytes_insert > 0) {
RecordTick(statistics_, BLOCK_CACHE_COMPRESSION_DICT_BYTES_INSERT,
get_context_stats_.num_cache_compression_dict_bytes_insert);
}
}
bool GetContext::SaveValue(const ParsedInternalKey& parsed_key,
const Slice& value, bool* matched,
Status* read_status, Cleanable* value_pinner) {
assert(matched);
assert((State() != kMerge && parsed_key.type != kTypeMerge) ||
merge_context_ != nullptr);
if (ucmp_->EqualWithoutTimestamp(parsed_key.user_key, user_key_)) {
*matched = true;
// If the value is not in the snapshot, skip it
if (!CheckCallback(parsed_key.sequence)) {
return true; // to continue to the next seq
}
if (seq_ != nullptr) {
// Set the sequence number if it is uninitialized
if (*seq_ == kMaxSequenceNumber) {
*seq_ = parsed_key.sequence;
}
if (max_covering_tombstone_seq_) {
*seq_ = std::max(*seq_, *max_covering_tombstone_seq_);
}
}
size_t ts_sz = ucmp_->timestamp_size();
Slice ts;
if (ts_sz > 0) {
// ensure always have ts if cf enables ts.
ts = ExtractTimestampFromUserKey(parsed_key.user_key, ts_sz);
if (timestamp_ != nullptr) {
if (!timestamp_->empty()) {
assert(ts_sz == timestamp_->size());
// `timestamp` can be set before `SaveValue` is ever called
// when max_covering_tombstone_seq_ was set.
// If this key has a higher sequence number than range tombstone,
// then timestamp should be updated. `ts_from_rangetombstone_` is
// set to false afterwards so that only the key with highest seqno
// updates the timestamp.
if (ts_from_rangetombstone_) {
assert(max_covering_tombstone_seq_);
if (parsed_key.sequence > *max_covering_tombstone_seq_) {
timestamp_->assign(ts.data(), ts.size());
ts_from_rangetombstone_ = false;
}
}
}
// TODO optimize for small size ts
const std::string kMaxTs(ts_sz, '\xff');
if (timestamp_->empty() ||
ucmp_->CompareTimestamp(*timestamp_, kMaxTs) == 0) {
timestamp_->assign(ts.data(), ts.size());
}
}
}
appendToReplayLog(parsed_key.type, value, ts);
auto type = parsed_key.type;
Slice unpacked_value = value;
// Key matches. Process it
if ((type == kTypeValue || type == kTypeValuePreferredSeqno ||
type == kTypeMerge || type == kTypeBlobIndex ||
type == kTypeWideColumnEntity || type == kTypeDeletion ||
type == kTypeDeletionWithTimestamp || type == kTypeSingleDeletion) &&
max_covering_tombstone_seq_ != nullptr &&
*max_covering_tombstone_seq_ > parsed_key.sequence) {
// Note that deletion types are also considered, this is for the case
// when we need to return timestamp to user. If a range tombstone has a
// higher seqno than point tombstone, its timestamp should be returned.
type = kTypeRangeDeletion;
}
switch (type) {
case kTypeValue:
case kTypeValuePreferredSeqno:
case kTypeBlobIndex:
case kTypeWideColumnEntity:
assert(State() == kNotFound || State() == kMerge);
if (type == kTypeValuePreferredSeqno) {
unpacked_value = ParsePackedValueForValue(value);
}
if (type == kTypeBlobIndex) {
if (is_blob_index_ == nullptr) {
// Blob value not supported. Stop.
state_ = kUnexpectedBlobIndex;
return false;
}
}
if (is_blob_index_ != nullptr) {
*is_blob_index_ = (type == kTypeBlobIndex);
}
if (State() == kNotFound) {
state_ = kFound;
if (do_merge_) {
if (type == kTypeBlobIndex && ucmp_->timestamp_size() != 0) {
ukey_with_ts_found_.PinSelf(parsed_key.user_key);
}
if (LIKELY(pinnable_val_ != nullptr)) {
Slice value_to_use = unpacked_value;
if (type == kTypeWideColumnEntity) {
const Status s = SaveWideColumnEntityToPinnable(
parsed_key.user_key, unpacked_value, value_pinner);
if (!s.ok()) {
if (s.IsIncomplete()) {
MarkKeyMayExist();
*read_status = s;
} else {
state_ = kCorrupt;
*read_status = s;
}
return false;
}
} else {
// Non-entity type
if (LIKELY(value_pinner != nullptr)) {
// If the backing resources for the value are provided, pin
// them
pinnable_val_->PinSlice(value_to_use, value_pinner);
} else {
TEST_SYNC_POINT_CALLBACK("GetContext::SaveValue::PinSelf",
this);
// Otherwise copy the value
pinnable_val_->PinSelf(value_to_use);
}
}
} else if (columns_ != nullptr) {
if (type == kTypeWideColumnEntity) {
const Status s = SaveWideColumnEntityToColumns(
parsed_key.user_key, unpacked_value, value_pinner);
if (!s.ok()) {
if (s.IsIncomplete()) {
MarkKeyMayExist();
*read_status = s;
} else {
state_ = kCorrupt;
*read_status = s;
}
return false;
}
} else {
columns_->SetPlainValue(unpacked_value, value_pinner);
}
}
} else {
// It means this function is called as part of DB GetMergeOperands
// API and the current value should be part of
// merge_context_->operand_list
if (type == kTypeBlobIndex) {
PinnableSlice pin_val;
if (GetBlobValue(parsed_key.user_key, unpacked_value, &pin_val,
read_status) == false) {
return false;
}
Slice blob_value(pin_val);
push_operand(blob_value, nullptr);
} else if (type == kTypeWideColumnEntity) {
const Status s = PushWideColumnEntityDefaultOperand(
parsed_key.user_key, unpacked_value, value_pinner);
if (!s.ok()) {
if (s.IsIncomplete()) {
MarkKeyMayExist();
*read_status = s;
} else {
state_ = kCorrupt;
*read_status = s;
}
return false;
}
} else {
assert(type == kTypeValue || type == kTypeValuePreferredSeqno);
push_operand(unpacked_value, value_pinner);
}
}
} else if (State() == kMerge) {
assert(merge_operator_ != nullptr);
if (type == kTypeBlobIndex) {
PinnableSlice pin_val;
if (GetBlobValue(parsed_key.user_key, unpacked_value, &pin_val,
read_status) == false) {
return false;
}
Slice blob_value(pin_val);
state_ = kFound;
if (do_merge_) {
const Status s = MergeWithPlainBaseValue(blob_value);
if (!s.ok()) {
*read_status = s;
return false;
}
} else {
// It means this function is called as part of DB GetMergeOperands
// API and the current value should be part of
// merge_context_->operand_list
push_operand(blob_value, nullptr);
}
} else if (type == kTypeWideColumnEntity) {
state_ = kFound;
if (do_merge_) {
const Status s = MergeWithWideColumnBaseValue(unpacked_value);
if (!s.ok()) {
*read_status = s;
return false;
}
} else {
// It means this function is called as part of DB GetMergeOperands
// API and the current value should be part of
// merge_context_->operand_list
const Status s = PushWideColumnEntityDefaultOperand(
parsed_key.user_key, unpacked_value, value_pinner);
if (!s.ok()) {
if (s.IsIncomplete()) {
MarkKeyMayExist();
*read_status = s;
} else {
state_ = kCorrupt;
*read_status = s;
}
return false;
}
}
} else {
assert(type == kTypeValue || type == kTypeValuePreferredSeqno);
state_ = kFound;
if (do_merge_) {
const Status s = MergeWithPlainBaseValue(unpacked_value);
if (!s.ok()) {
*read_status = s;
return false;
}
} else {
// It means this function is called as part of DB GetMergeOperands
// API and the current value should be part of
// merge_context_->operand_list
push_operand(unpacked_value, value_pinner);
}
}
}
return false;
case kTypeDeletion:
case kTypeDeletionWithTimestamp:
case kTypeSingleDeletion:
case kTypeRangeDeletion:
// TODO(noetzli): Verify correctness once merge of single-deletes
// is supported
assert(State() == kNotFound || State() == kMerge);
if (State() == kNotFound) {
state_ = kDeleted;
} else if (State() == kMerge) {
state_ = kFound;
if (do_merge_) {
const Status s = MergeWithNoBaseValue();
if (!s.ok()) {
*read_status = s;
return false;
}
}
// If do_merge_ = false then the current value shouldn't be part of
// merge_context_->operand_list
}
return false;
case kTypeMerge:
assert(State() == kNotFound || State() == kMerge);
state_ = kMerge;
// value_pinner is not set from plain_table_reader.cc for example.
push_operand(value, value_pinner);
PERF_COUNTER_ADD(internal_merge_point_lookup_count, 1);
if (do_merge_ && merge_operator_ != nullptr &&
merge_operator_->ShouldMerge(
merge_context_->GetOperandsDirectionBackward())) {
state_ = kFound;
const Status s = MergeWithNoBaseValue();
if (!s.ok()) {
*read_status = s;
return false;
}
return false;
}
if (merge_context_->get_merge_operands_options != nullptr &&
merge_context_->get_merge_operands_options->continue_cb !=
nullptr &&
!merge_context_->get_merge_operands_options->continue_cb(value)) {
state_ = kFound;
return false;
}
return true;
default:
assert(false);
break;
}
}
// State() could be Corrupt, merge or notfound
return false;
}
Status GetContext::PostprocessMerge(const Status& merge_status) {
if (!merge_status.ok()) {
if (merge_status.subcode() == Status::SubCode::kMergeOperatorFailed) {
state_ = kMergeOperatorFailed;
} else {
state_ = kCorrupt;
}
return merge_status;
}
if (LIKELY(pinnable_val_ != nullptr)) {
pinnable_val_->PinSelf();
}
return Status::OK();
}
Status GetContext::MergeWithNoBaseValue() {
assert(do_merge_);
assert(pinnable_val_ || columns_);
assert(!pinnable_val_ || !columns_);
// `op_failure_scope` (an output parameter) is not provided (set to nullptr)
// since a failure must be propagated regardless of its value.
const Status s = MergeHelper::TimedFullMerge(
merge_operator_, user_key_, MergeHelper::kNoBaseValue,
merge_context_->GetOperands(), logger_, statistics_, clock_,
/* update_num_ops_stats */ true, /* op_failure_scope */ nullptr,
pinnable_val_ ? pinnable_val_->GetSelf() : nullptr, columns_);
return PostprocessMerge(s);
}
Status GetContext::MergeWithPlainBaseValue(const Slice& value) {
assert(do_merge_);
assert(pinnable_val_ || columns_);
assert(!pinnable_val_ || !columns_);
// `op_failure_scope` (an output parameter) is not provided (set to nullptr)
// since a failure must be propagated regardless of its value.
const Status s = MergeHelper::TimedFullMerge(
merge_operator_, user_key_, MergeHelper::kPlainBaseValue, value,
merge_context_->GetOperands(), logger_, statistics_, clock_,
/* update_num_ops_stats */ true, /* op_failure_scope */ nullptr,
pinnable_val_ ? pinnable_val_->GetSelf() : nullptr, columns_);
return PostprocessMerge(s);
}
Status GetContext::MergeWithWideColumnBaseValue(const Slice& entity) {
assert(do_merge_);
assert(pinnable_val_ || columns_);
assert(!pinnable_val_ || !columns_);
// Resolve V2 entity blob columns if present, since TimedFullMerge only
// supports V1 format.
std::string resolved_entity;
Slice effective_entity;
Status s_resolve = WideColumnSerialization::ResolveEntityForMerge(
entity, user_key_, blob_fetcher_, nullptr /* prefetch_buffers */,
resolved_entity, effective_entity);
if (!s_resolve.ok()) {
if (s_resolve.IsIncomplete()) {
MarkKeyMayExist();
return s_resolve;
}
state_ = kCorrupt;
return s_resolve;
}
// `op_failure_scope` (an output parameter) is not provided (set to nullptr)
// since a failure must be propagated regardless of its value.
const Status s = MergeHelper::TimedFullMerge(
merge_operator_, user_key_, MergeHelper::kWideBaseValue, effective_entity,
merge_context_->GetOperands(), logger_, statistics_, clock_,
/* update_num_ops_stats */ true, /* op_failure_scope */ nullptr,
pinnable_val_ ? pinnable_val_->GetSelf() : nullptr, columns_);
return PostprocessMerge(s);
}
bool GetContext::GetBlobValue(const Slice& user_key, const Slice& blob_index,
PinnableSlice* blob_value, Status* read_status) {
constexpr FilePrefetchBuffer* prefetch_buffer = nullptr;
constexpr uint64_t* bytes_read = nullptr;
*read_status = blob_fetcher_->FetchBlob(user_key, blob_index, prefetch_buffer,
blob_value, bytes_read);
if (!read_status->ok()) {
if (read_status->IsIncomplete()) {
// FIXME: this code is not covered by unit tests
MarkKeyMayExist();
return false;
}
state_ = kCorrupt;
return false;
}
*is_blob_index_ = false;
return true;
}
void GetContext::push_operand(const Slice& value, Cleanable* value_pinner) {
// TODO(yanqin) preserve timestamps information in merge_context
if (pinned_iters_mgr() && pinned_iters_mgr()->PinningEnabled() &&
value_pinner != nullptr) {
value_pinner->DelegateCleanupsTo(pinned_iters_mgr());
merge_context_->PushOperand(value, true /*value_pinned*/);
} else {
merge_context_->PushOperand(value, false);
}
}
Status replayGetContextLog(const Slice& replay_log, const Slice& user_key,
GetContext* get_context, Cleanable* value_pinner,
SequenceNumber seq_no) {
Slice s = replay_log;
Slice ts;
size_t ts_sz = get_context->TimestampSize();
bool ret = false;
while (s.size()) {
auto type = static_cast<ValueType>(*s.data());
s.remove_prefix(1);
Slice value;
ret = GetLengthPrefixedSlice(&s, &value);
assert(ret);
bool dont_care __attribute__((__unused__));
// Use a copy to prevent modifying user_key. Modification of user_key
// could result to potential cache miss.
std::string user_key_str = user_key.ToString();
ParsedInternalKey ikey = ParsedInternalKey(user_key_str, seq_no, type);
// If ts enabled for current cf, there will always be ts appended after each
// piece of value.
if (ts_sz > 0) {
ret = GetLengthPrefixedSlice(&s, &ts);
assert(ts_sz == ts.size());
assert(ret);
ikey.SetTimestamp(ts);
}
(void)ret;
Status read_status = Status::OK();
const bool keep_replaying = get_context->SaveValue(
ikey, value, &dont_care, &read_status, value_pinner);
if (!read_status.ok()) {
return read_status;
}
if (!keep_replaying) {
// SaveValue() reached a terminal state for this row-cache replay.
break;
}
}
return Status::OK();
}
} // namespace ROCKSDB_NAMESPACE
Reference in New Issue View Git Blame Copy Permalink