dolysis
/
rocksdb
mirror of https://github.com/facebook/rocksdb.git
3
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity
rocksdb/db/merge_helper.cc

610 lines
23 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 "db/merge_helper.h"
#include <string>
#include "db/blob/blob_fetcher.h"
#include "db/blob/blob_index.h"
#include "db/blob/prefetch_buffer_collection.h"
#include "db/compaction/compaction_iteration_stats.h"
#include "db/dbformat.h"
#include "db/wide/wide_column_serialization.h"
#include "logging/logging.h"
#include "monitoring/perf_context_imp.h"
#include "monitoring/statistics.h"
#include "port/likely.h"
#include "rocksdb/comparator.h"
#include "rocksdb/db.h"
#include "rocksdb/merge_operator.h"
#include "rocksdb/system_clock.h"
#include "table/format.h"
#include "table/internal_iterator.h"
namespace ROCKSDB_NAMESPACE {
MergeHelper::MergeHelper(Env* env, const Comparator* user_comparator,
const MergeOperator* user_merge_operator,
const CompactionFilter* compaction_filter,
Logger* logger, bool assert_valid_internal_key,
SequenceNumber latest_snapshot,
const SnapshotChecker* snapshot_checker, int level,
Statistics* stats,
const std::atomic<bool>* shutting_down)
: env_(env),
clock_(env->GetSystemClock().get()),
user_comparator_(user_comparator),
user_merge_operator_(user_merge_operator),
compaction_filter_(compaction_filter),
shutting_down_(shutting_down),
logger_(logger),
assert_valid_internal_key_(assert_valid_internal_key),
allow_single_operand_(false),
latest_snapshot_(latest_snapshot),
snapshot_checker_(snapshot_checker),
level_(level),
keys_(),
filter_timer_(clock_),
total_filter_time_(0U),
stats_(stats) {
assert(user_comparator_ != nullptr);
if (user_merge_operator_) {
allow_single_operand_ = user_merge_operator_->AllowSingleOperand();
}
}
Status MergeHelper::TimedFullMerge(const MergeOperator* merge_operator,
const Slice& key, const Slice* value,
const std::vector<Slice>& operands,
std::string* result, Logger* logger,
Statistics* statistics, SystemClock* clock,
Slice* result_operand,
bool update_num_ops_stats) {
assert(merge_operator != nullptr);
if (operands.empty()) {
assert(value != nullptr && result != nullptr);
result->assign(value->data(), value->size());
return Status::OK();
}
if (update_num_ops_stats) {
RecordInHistogram(statistics, READ_NUM_MERGE_OPERANDS,
static_cast<uint64_t>(operands.size()));
}
bool success = false;
Slice tmp_result_operand(nullptr, 0);
const MergeOperator::MergeOperationInput merge_in(key, value, operands,
logger);
MergeOperator::MergeOperationOutput merge_out(*result, tmp_result_operand);
{
// Setup to time the merge
StopWatchNano timer(clock, statistics != nullptr);
PERF_TIMER_GUARD(merge_operator_time_nanos);
// Do the merge
success = merge_operator->FullMergeV2(merge_in, &merge_out);
if (tmp_result_operand.data()) {
// FullMergeV2 result is an existing operand
if (result_operand != nullptr) {
*result_operand = tmp_result_operand;
} else {
result->assign(tmp_result_operand.data(), tmp_result_operand.size());
}
} else if (result_operand) {
*result_operand = Slice(nullptr, 0);
}
RecordTick(statistics, MERGE_OPERATION_TOTAL_TIME,
statistics ? timer.ElapsedNanos() : 0);
}
if (!success) {
RecordTick(statistics, NUMBER_MERGE_FAILURES);
return Status::Corruption("Error: Could not perform merge.");
}
return Status::OK();
}
Status MergeHelper::TimedFullMerge(const MergeOperator* merge_operator,
const Slice& key, const Slice* base_value,
const std::vector<Slice>& operands,
std::string* value,
PinnableWideColumns* columns, Logger* logger,
Statistics* statistics, SystemClock* clock,
Slice* result_operand,
bool update_num_ops_stats) {
assert(value || columns);
assert(!value || !columns);
std::string result;
const Status s =
TimedFullMerge(merge_operator, key, base_value, operands, &result, logger,
statistics, clock, result_operand, update_num_ops_stats);
if (!s.ok()) {
return s;
}
if (value) {
*value = std::move(result);
return Status::OK();
}
assert(columns);
columns->SetPlainValue(result);
return Status::OK();
}
Status MergeHelper::TimedFullMergeWithEntity(
const MergeOperator* merge_operator, const Slice& key, Slice base_entity,
const std::vector<Slice>& operands, std::string* value,
PinnableWideColumns* columns, Logger* logger, Statistics* statistics,
SystemClock* clock, Slice* result_operand, bool update_num_ops_stats) {
assert(value || columns);
assert(!value || !columns);
WideColumns base_columns;
{
const Status s =
WideColumnSerialization::Deserialize(base_entity, base_columns);
if (!s.ok()) {
return s;
}
}
const bool has_default_column =
!base_columns.empty() && base_columns[0].name() == kDefaultWideColumnName;
Slice value_of_default;
if (has_default_column) {
value_of_default = base_columns[0].value();
}
std::string result;
{
const Status s = TimedFullMerge(
merge_operator, key, &value_of_default, operands, &result, logger,
statistics, clock, result_operand, update_num_ops_stats);
if (!s.ok()) {
return s;
}
}
if (value) {
*value = std::move(result);
return Status::OK();
}
assert(columns);
std::string output;
if (has_default_column) {
base_columns[0].value() = result;
const Status s = WideColumnSerialization::Serialize(base_columns, output);
if (!s.ok()) {
return s;
}
} else {
const Status s =
WideColumnSerialization::Serialize(result, base_columns, output);
if (!s.ok()) {
return s;
}
}
return columns->SetWideColumnValue(output);
}
// PRE: iter points to the first merge type entry
// POST: iter points to the first entry beyond the merge process (or the end)
// keys_, operands_ are updated to reflect the merge result.
// keys_ stores the list of keys encountered while merging.
// operands_ stores the list of merge operands encountered while merging.
// keys_[i] corresponds to operands_[i] for each i.
//
// TODO: Avoid the snapshot stripe map lookup in CompactionRangeDelAggregator
// and just pass the StripeRep corresponding to the stripe being merged.
Status MergeHelper::MergeUntil(InternalIterator* iter,
CompactionRangeDelAggregator* range_del_agg,
const SequenceNumber stop_before,
const bool at_bottom,
const bool allow_data_in_errors,
const BlobFetcher* blob_fetcher,
const std::string* const full_history_ts_low,
PrefetchBufferCollection* prefetch_buffers,
CompactionIterationStats* c_iter_stats) {
// Get a copy of the internal key, before it's invalidated by iter->Next()
// Also maintain the list of merge operands seen.
assert(HasOperator());
keys_.clear();
merge_context_.Clear();
has_compaction_filter_skip_until_ = false;
assert(user_merge_operator_);
assert(user_comparator_);
const size_t ts_sz = user_comparator_->timestamp_size();
if (full_history_ts_low) {
assert(ts_sz > 0);
assert(ts_sz == full_history_ts_low->size());
}
bool first_key = true;
// We need to parse the internal key again as the parsed key is
// backed by the internal key!
// Assume no internal key corruption as it has been successfully parsed
// by the caller.
// original_key_is_iter variable is just caching the information:
// original_key_is_iter == (iter->key().ToString() == original_key)
bool original_key_is_iter = true;
std::string original_key = iter->key().ToString();
// Important:
// orig_ikey is backed by original_key if keys_.empty()
// orig_ikey is backed by keys_.back() if !keys_.empty()
ParsedInternalKey orig_ikey;
Status s = ParseInternalKey(original_key, &orig_ikey, allow_data_in_errors);
assert(s.ok());
if (!s.ok()) return s;
assert(kTypeMerge == orig_ikey.type);
bool hit_the_next_user_key = false;
int cmp_with_full_history_ts_low = 0;
for (; iter->Valid(); iter->Next(), original_key_is_iter = false) {
if (IsShuttingDown()) {
s = Status::ShutdownInProgress();
return s;
}
ParsedInternalKey ikey;
assert(keys_.size() == merge_context_.GetNumOperands());
Status pik_status =
ParseInternalKey(iter->key(), &ikey, allow_data_in_errors);
Slice ts;
if (pik_status.ok()) {
ts = ExtractTimestampFromUserKey(ikey.user_key, ts_sz);
if (full_history_ts_low) {
cmp_with_full_history_ts_low =
user_comparator_->CompareTimestamp(ts, *full_history_ts_low);
}
}
if (!pik_status.ok()) {
// stop at corrupted key
if (assert_valid_internal_key_) {
return pik_status;
}
break;
} else if (first_key) {
// If user-defined timestamp is enabled, we expect both user key and
// timestamps are equal, as a sanity check.
assert(user_comparator_->Equal(ikey.user_key, orig_ikey.user_key));
first_key = false;
} else if (!user_comparator_->EqualWithoutTimestamp(ikey.user_key,
orig_ikey.user_key) ||
(ts_sz > 0 &&
!user_comparator_->Equal(ikey.user_key, orig_ikey.user_key) &&
cmp_with_full_history_ts_low >= 0)) {
// 1) hit a different user key, or
// 2) user-defined timestamp is enabled, and hit a version of user key NOT
// eligible for GC, then stop right here.
hit_the_next_user_key = true;
break;
} else if (stop_before > 0 && ikey.sequence <= stop_before &&
LIKELY(snapshot_checker_ == nullptr ||
snapshot_checker_->CheckInSnapshot(ikey.sequence,
stop_before) !=
SnapshotCheckerResult::kNotInSnapshot)) {
// hit an entry that's possibly visible by the previous snapshot, can't
// touch that
break;
}
// At this point we are guaranteed that we need to process this key.
assert(IsValueType(ikey.type));
if (ikey.type != kTypeMerge) {
// hit a put/delete/single delete
// => merge the put value or a nullptr with operands_
// => store result in operands_.back() (and update keys_.back())
// => change the entry type to kTypeValue for keys_.back()
// We are done! Success!
// If there are no operands, just return the Status::OK(). That will cause
// the compaction iterator to write out the key we're currently at, which
// is the put/delete we just encountered.
if (keys_.empty()) {
return s;
}
// TODO(noetzli) If the merge operator returns false, we are currently
// (almost) silently dropping the put/delete. That's probably not what we
// want. Also if we're in compaction and it's a put, it would be nice to
// run compaction filter on it.
const Slice val = iter->value();
PinnableSlice blob_value;
const Slice* val_ptr;
if ((kTypeValue == ikey.type || kTypeBlobIndex == ikey.type ||
kTypeWideColumnEntity == ikey.type) &&
(range_del_agg == nullptr ||
!range_del_agg->ShouldDelete(
ikey, RangeDelPositioningMode::kForwardTraversal))) {
if (ikey.type == kTypeWideColumnEntity) {
// TODO: support wide-column entities
return Status::NotSupported(
"Merge currently not supported for wide-column entities");
} else if (ikey.type == kTypeBlobIndex) {
BlobIndex blob_index;
s = blob_index.DecodeFrom(val);
if (!s.ok()) {
return s;
}
FilePrefetchBuffer* prefetch_buffer =
prefetch_buffers ? prefetch_buffers->GetOrCreatePrefetchBuffer(
blob_index.file_number())
: nullptr;
uint64_t bytes_read = 0;
assert(blob_fetcher);
s = blob_fetcher->FetchBlob(ikey.user_key, blob_index,
prefetch_buffer, &blob_value,
&bytes_read);
if (!s.ok()) {
return s;
}
val_ptr = &blob_value;
if (c_iter_stats) {
++c_iter_stats->num_blobs_read;
c_iter_stats->total_blob_bytes_read += bytes_read;
}
} else {
val_ptr = &val;
}
} else {
val_ptr = nullptr;
}
std::string merge_result;
s = TimedFullMerge(user_merge_operator_, ikey.user_key, val_ptr,
merge_context_.GetOperands(), &merge_result, logger_,
stats_, clock_);
// We store the result in keys_.back() and operands_.back()
// if nothing went wrong (i.e.: no operand corruption on disk)
if (s.ok()) {
// The original key encountered
original_key = std::move(keys_.back());
orig_ikey.type = kTypeValue;
UpdateInternalKey(&original_key, orig_ikey.sequence, orig_ikey.type);
keys_.clear();
merge_context_.Clear();
keys_.emplace_front(std::move(original_key));
merge_context_.PushOperand(merge_result);
}
// move iter to the next entry
iter->Next();
return s;
} else {
// hit a merge
// => if there is a compaction filter, apply it.
// => check for range tombstones covering the operand
// => merge the operand into the front of the operands_ list
// if not filtered
// => then continue because we haven't yet seen a Put/Delete.
//
// Keep queuing keys and operands until we either meet a put / delete
// request or later did a partial merge.
Slice value_slice = iter->value();
// add an operand to the list if:
// 1) it's included in one of the snapshots. in that case we *must* write
// it out, no matter what compaction filter says
// 2) it's not filtered by a compaction filter
CompactionFilter::Decision filter =
ikey.sequence <= latest_snapshot_
? CompactionFilter::Decision::kKeep
: FilterMerge(orig_ikey.user_key, value_slice);
if (filter != CompactionFilter::Decision::kRemoveAndSkipUntil &&
range_del_agg != nullptr &&
range_del_agg->ShouldDelete(
iter->key(), RangeDelPositioningMode::kForwardTraversal)) {
filter = CompactionFilter::Decision::kRemove;
}
if (filter == CompactionFilter::Decision::kKeep ||
filter == CompactionFilter::Decision::kChangeValue) {
if (original_key_is_iter) {
// this is just an optimization that saves us one memcpy
keys_.emplace_front(original_key);
} else {
keys_.emplace_front(iter->key().ToString());
}
if (keys_.size() == 1) {
// we need to re-anchor the orig_ikey because it was anchored by
// original_key before
pik_status =
ParseInternalKey(keys_.back(), &orig_ikey, allow_data_in_errors);
pik_status.PermitUncheckedError();
assert(pik_status.ok());
}
if (filter == CompactionFilter::Decision::kKeep) {
merge_context_.PushOperand(
value_slice, iter->IsValuePinned() /* operand_pinned */);
} else {
assert(filter == CompactionFilter::Decision::kChangeValue);
// Compaction filter asked us to change the operand from value_slice
// to compaction_filter_value_.
merge_context_.PushOperand(compaction_filter_value_, false);
}
} else if (filter == CompactionFilter::Decision::kRemoveAndSkipUntil) {
// Compaction filter asked us to remove this key altogether
// (not just this operand), along with some keys following it.
keys_.clear();
merge_context_.Clear();
has_compaction_filter_skip_until_ = true;
return s;
}
}
}
if (cmp_with_full_history_ts_low >= 0) {
size_t num_merge_operands = merge_context_.GetNumOperands();
if (ts_sz && num_merge_operands > 1) {
// We do not merge merge operands with different timestamps if they are
// not eligible for GC.
ROCKS_LOG_ERROR(logger_, "ts_sz=%d, %d merge oprands",
static_cast<int>(ts_sz),
static_cast<int>(num_merge_operands));
assert(false);
}
}
if (merge_context_.GetNumOperands() == 0) {
// we filtered out all the merge operands
return s;
}
// We are sure we have seen this key's entire history if:
// at_bottom == true (this does not necessarily mean it is the bottommost
// layer, but rather that we are confident the key does not appear on any of
// the lower layers, at_bottom == false doesn't mean it does appear, just
// that we can't be sure, see Compaction::IsBottommostLevel for details)
// AND
// we have either encountered another key or end of key history on this
// layer.
// Note that if user-defined timestamp is enabled, we need some extra caution
// here: if full_history_ts_low is nullptr, or it's not null but the key's
// timestamp is greater than or equal to full_history_ts_low, it means this
// key cannot be dropped. We may not have seen the beginning of the key.
//
// When these conditions are true we are able to merge all the keys
// using full merge.
//
// For these cases we are not sure about, we simply miss the opportunity
// to combine the keys. Since VersionSet::SetupOtherInputs() always makes
// sure that all merge-operands on the same level get compacted together,
// this will simply lead to these merge operands moving to the next level.
bool surely_seen_the_beginning =
(hit_the_next_user_key || !iter->Valid()) && at_bottom &&
(ts_sz == 0 || cmp_with_full_history_ts_low < 0);
if (surely_seen_the_beginning) {
// do a final merge with nullptr as the existing value and say
// bye to the merge type (it's now converted to a Put)
assert(kTypeMerge == orig_ikey.type);
assert(merge_context_.GetNumOperands() >= 1);
assert(merge_context_.GetNumOperands() == keys_.size());
std::string merge_result;
s = TimedFullMerge(user_merge_operator_, orig_ikey.user_key, nullptr,
merge_context_.GetOperands(), &merge_result, logger_,
stats_, clock_);
if (s.ok()) {
// The original key encountered
// We are certain that keys_ is not empty here (see assertions couple of
// lines before).
original_key = std::move(keys_.back());
orig_ikey.type = kTypeValue;
UpdateInternalKey(&original_key, orig_ikey.sequence, orig_ikey.type);
keys_.clear();
merge_context_.Clear();
keys_.emplace_front(std::move(original_key));
merge_context_.PushOperand(merge_result);
}
} else {
// We haven't seen the beginning of the key nor a Put/Delete.
// Attempt to use the user's associative merge function to
// merge the stacked merge operands into a single operand.
s = Status::MergeInProgress();
if (merge_context_.GetNumOperands() >= 2 ||
(allow_single_operand_ && merge_context_.GetNumOperands() == 1)) {
bool merge_success = false;
std::string merge_result;
{
StopWatchNano timer(clock_, stats_ != nullptr);
PERF_TIMER_GUARD(merge_operator_time_nanos);
merge_success = user_merge_operator_->PartialMergeMulti(
orig_ikey.user_key,
std::deque<Slice>(merge_context_.GetOperands().begin(),
merge_context_.GetOperands().end()),
&merge_result, logger_);
RecordTick(stats_, MERGE_OPERATION_TOTAL_TIME,
stats_ ? timer.ElapsedNanosSafe() : 0);
}
if (merge_success) {
// Merging of operands (associative merge) was successful.
// Replace operands with the merge result
merge_context_.Clear();
merge_context_.PushOperand(merge_result);
keys_.erase(keys_.begin(), keys_.end() - 1);
}
}
}
return s;
}
MergeOutputIterator::MergeOutputIterator(const MergeHelper* merge_helper)
: merge_helper_(merge_helper) {
it_keys_ = merge_helper_->keys().rend();
it_values_ = merge_helper_->values().rend();
}
void MergeOutputIterator::SeekToFirst() {
const auto& keys = merge_helper_->keys();
const auto& values = merge_helper_->values();
assert(keys.size() == values.size());
it_keys_ = keys.rbegin();
it_values_ = values.rbegin();
}
void MergeOutputIterator::Next() {
++it_keys_;
++it_values_;
}
CompactionFilter::Decision MergeHelper::FilterMerge(const Slice& user_key,
const Slice& value_slice) {
if (compaction_filter_ == nullptr) {
return CompactionFilter::Decision::kKeep;
}
if (stats_ != nullptr && ShouldReportDetailedTime(env_, stats_)) {
filter_timer_.Start();
}
compaction_filter_value_.clear();
compaction_filter_skip_until_.Clear();
auto ret = compaction_filter_->FilterV2(
level_, user_key, CompactionFilter::ValueType::kMergeOperand, value_slice,
&compaction_filter_value_, compaction_filter_skip_until_.rep());
if (ret == CompactionFilter::Decision::kRemoveAndSkipUntil) {
if (user_comparator_->Compare(*compaction_filter_skip_until_.rep(),
user_key) <= 0) {
// Invalid skip_until returned from compaction filter.
// Keep the key as per FilterV2 documentation.
ret = CompactionFilter::Decision::kKeep;
} else {
compaction_filter_skip_until_.ConvertFromUserKey(kMaxSequenceNumber,
kValueTypeForSeek);
}
}
if (stats_ != nullptr && ShouldReportDetailedTime(env_, stats_)) {
total_filter_time_ += filter_timer_.ElapsedNanosSafe();
}
return ret;
}
} // namespace ROCKSDB_NAMESPACE
Reference in New Issue View Git Blame Copy Permalink