From fbd9077d66776828dc466daa65a91d74192f1ec8 Mon Sep 17 00:00:00 2001 From: Levi Tamasi Date: Mon, 7 Nov 2022 15:42:10 -0800 Subject: [PATCH] Fix a bug where GetContext does not update READ_NUM_MERGE_OPERANDS (#10925) Summary: The patch fixes a bug where `GetContext::Merge` (and `MergeEntity`) does not update the ticker `READ_NUM_MERGE_OPERANDS` because it implicitly uses the default parameter value of `update_num_ops_stats=false` when calling `MergeHelper::TimedFullMerge`. Also, to prevent such issues going forward, the PR removes the default parameter values from the `TimedFullMerge` methods. In addition, it removes an unused/unnecessary parameter from `TimedFullMergeWithEntity`, and does some cleanup at the call sites of these methods. Pull Request resolved: https://github.com/facebook/rocksdb/pull/10925 Test Plan: `make check` Reviewed By: riversand963 Differential Revision: D41096453 Pulled By: ltamasi fbshipit-source-id: fc60646d32b4d516b8fe81e265c3f020a32fd7f8 --- HISTORY.md | 1 + db/db_iter.cc | 3 ++- db/memtable.cc | 11 +++++++---- db/merge_helper.cc | 18 +++++++++++------- db/merge_helper.h | 12 +++++------- db/version_set.cc | 4 ++-- db/write_batch.cc | 3 ++- table/get_context.cc | 5 +++-- .../write_batch_with_index_internal.cc | 17 ++++++++++------- 9 files changed, 43 insertions(+), 31 deletions(-) diff --git a/HISTORY.md b/HISTORY.md index 33f87d36d..207d22f26 100644 --- a/HISTORY.md +++ b/HISTORY.md @@ -7,6 +7,7 @@ * Fix FIFO compaction causing corruption of overlapping seqnos in L0 files due to ingesting files of overlapping seqnos with memtable's under `CompactionOptionsFIFO::allow_compaction=true` or `CompactionOptionsFIFO::age_for_warm>0` or `CompactRange()/CompactFiles()` is used. Before the fix, `force_consistency_checks=true` may catch the corruption before it's exposed to readers, in which case writes returning `Status::Corruption` would be expected. * Fix memory corruption error in scans if async_io is enabled. Memory corruption happened if there is IOError while reading the data leading to empty buffer and other buffer already in progress of async read goes again for reading. * Fix failed memtable flush retry bug that could cause wrongly ordered updates, which would surface to writers as `Status::Corruption` in case of `force_consistency_checks=true` (default). It affects use cases that enable both parallel flush (`max_background_flushes > 1` or `max_background_jobs >= 8`) and non-default memtable count (`max_write_buffer_number > 2`). +* Fixed an issue where the `READ_NUM_MERGE_OPERANDS` ticker was not updated when the base key-value or tombstone was read from an SST file. ### New Features * Add basic support for user-defined timestamp to Merge (#10819). diff --git a/db/db_iter.cc b/db/db_iter.cc index d0a6698a5..6f8319910 100644 --- a/db/db_iter.cc +++ b/db/db_iter.cc @@ -1247,7 +1247,8 @@ bool DBIter::FindValueForCurrentKeyUsingSeek() { Status DBIter::Merge(const Slice* val, const Slice& user_key) { Status s = MergeHelper::TimedFullMerge( merge_operator_, user_key, val, merge_context_.GetOperands(), - &saved_value_, logger_, statistics_, clock_, &pinned_value_, true); + &saved_value_, logger_, statistics_, clock_, &pinned_value_, + /* update_num_ops_stats */ true); if (!s.ok()) { valid_ = false; status_ = s; diff --git a/db/memtable.cc b/db/memtable.cc index 7328f9846..829a3c099 100644 --- a/db/memtable.cc +++ b/db/memtable.cc @@ -1069,7 +1069,8 @@ static bool SaveValue(void* arg, const char* entry) { *(s->status) = MergeHelper::TimedFullMerge( merge_operator, s->key->user_key(), &v, merge_context->GetOperands(), s->value, s->columns, s->logger, - s->statistics, s->clock, nullptr /* result_operand */, true); + s->statistics, s->clock, /* result_operand */ nullptr, + /* update_num_ops_stats */ true); } } else if (s->value) { s->value->assign(v.data(), v.size()); @@ -1118,7 +1119,7 @@ static bool SaveValue(void* arg, const char* entry) { *(s->status) = MergeHelper::TimedFullMergeWithEntity( merge_operator, s->key->user_key(), v, merge_context->GetOperands(), s->value, s->columns, s->logger, - s->statistics, s->clock, nullptr /* result_operand */, true); + s->statistics, s->clock, /* update_num_ops_stats */ true); } } else if (s->value) { Slice value_of_default; @@ -1152,7 +1153,8 @@ static bool SaveValue(void* arg, const char* entry) { *(s->status) = MergeHelper::TimedFullMerge( merge_operator, s->key->user_key(), nullptr, merge_context->GetOperands(), s->value, s->columns, s->logger, - s->statistics, s->clock, nullptr /* result_operand */, true); + s->statistics, s->clock, /* result_operand */ nullptr, + /* update_num_ops_stats */ true); } } else { *(s->status) = Status::NotFound(); @@ -1181,7 +1183,8 @@ static bool SaveValue(void* arg, const char* entry) { *(s->status) = MergeHelper::TimedFullMerge( merge_operator, s->key->user_key(), nullptr, merge_context->GetOperands(), s->value, s->columns, s->logger, - s->statistics, s->clock, nullptr /* result_operand */, true); + s->statistics, s->clock, /* result_operand */ nullptr, + /* update_num_ops_stats */ true); } *(s->found_final_value) = true; diff --git a/db/merge_helper.cc b/db/merge_helper.cc index 5ece49616..5a7c5765e 100644 --- a/db/merge_helper.cc +++ b/db/merge_helper.cc @@ -146,7 +146,7 @@ Status MergeHelper::TimedFullMergeWithEntity( const MergeOperator* merge_operator, const Slice& key, Slice base_entity, const std::vector& operands, std::string* value, PinnableWideColumns* columns, Logger* logger, Statistics* statistics, - SystemClock* clock, Slice* result_operand, bool update_num_ops_stats) { + SystemClock* clock, bool update_num_ops_stats) { assert(value || columns); assert(!value || !columns); @@ -171,6 +171,8 @@ Status MergeHelper::TimedFullMergeWithEntity( std::string result; { + constexpr Slice* result_operand = nullptr; + const Status s = TimedFullMerge( merge_operator, key, &value_of_default, operands, &result, logger, statistics, clock, result_operand, update_num_ops_stats); @@ -380,9 +382,10 @@ Status MergeHelper::MergeUntil(InternalIterator* iter, 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_); + s = TimedFullMerge( + user_merge_operator_, ikey.user_key, val_ptr, + merge_context_.GetOperands(), &merge_result, logger_, stats_, clock_, + /* result_operand */ nullptr, /* update_num_ops_stats */ false); // We store the result in keys_.back() and operands_.back() // if nothing went wrong (i.e.: no operand corruption on disk) @@ -509,9 +512,10 @@ Status MergeHelper::MergeUntil(InternalIterator* iter, 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_); + s = TimedFullMerge( + user_merge_operator_, orig_ikey.user_key, nullptr, + merge_context_.GetOperands(), &merge_result, logger_, stats_, clock_, + /* result_operand */ nullptr, /* update_num_ops_stats */ false); if (s.ok()) { // The original key encountered // We are certain that keys_ is not empty here (see assertions couple of diff --git a/db/merge_helper.h b/db/merge_helper.h index 4b6328d24..923850a08 100644 --- a/db/merge_helper.h +++ b/db/merge_helper.h @@ -54,24 +54,22 @@ class MergeHelper { const std::vector& operands, std::string* result, Logger* logger, Statistics* statistics, SystemClock* clock, - Slice* result_operand = nullptr, - bool update_num_ops_stats = false); + Slice* result_operand, + bool update_num_ops_stats); static Status TimedFullMerge(const MergeOperator* merge_operator, const Slice& key, const Slice* base_value, const std::vector& operands, std::string* value, PinnableWideColumns* columns, Logger* logger, Statistics* statistics, - SystemClock* clock, - Slice* result_operand = nullptr, - bool update_num_ops_stats = false); + SystemClock* clock, Slice* result_operand, + bool update_num_ops_stats); static Status TimedFullMergeWithEntity( const MergeOperator* merge_operator, const Slice& key, Slice base_entity, const std::vector& operands, std::string* value, PinnableWideColumns* columns, Logger* logger, Statistics* statistics, - SystemClock* clock, Slice* result_operand = nullptr, - bool update_num_ops_stats = false); + SystemClock* clock, bool update_num_ops_stats); // During compaction, merge entries until we hit // - a corrupted key diff --git a/db/version_set.cc b/db/version_set.cc index 4a2043263..aa0dc394e 100644 --- a/db/version_set.cc +++ b/db/version_set.cc @@ -2389,7 +2389,7 @@ void Version::Get(const ReadOptions& read_options, const LookupKey& k, *status = MergeHelper::TimedFullMerge( merge_operator_, user_key, nullptr, merge_context->GetOperands(), str_value, columns, info_log_, db_statistics_, clock_, - nullptr /* result_operand */, true); + /* result_operand */ nullptr, /* update_num_ops_stats */ true); if (status->ok()) { if (LIKELY(value != nullptr)) { value->PinSelf(); @@ -2630,7 +2630,7 @@ void Version::MultiGet(const ReadOptions& read_options, MultiGetRange* range, *status = MergeHelper::TimedFullMerge( merge_operator_, user_key, nullptr, iter->merge_context.GetOperands(), str_value, info_log_, db_statistics_, clock_, - nullptr /* result_operand */, true); + /* result_operand */ nullptr, /* update_num_ops_stats */ true); if (LIKELY(iter->value != nullptr)) { iter->value->PinSelf(); range->AddValueSize(iter->value->size()); diff --git a/db/write_batch.cc b/db/write_batch.cc index c5042acf0..796697cfc 100644 --- a/db/write_batch.cc +++ b/db/write_batch.cc @@ -2502,7 +2502,8 @@ class MemTableInserter : public WriteBatch::Handler { Status merge_status = MergeHelper::TimedFullMerge( merge_operator, key, &get_value_slice, {value}, &new_value, moptions->info_log, moptions->statistics, - SystemClock::Default().get()); + SystemClock::Default().get(), /* result_operand */ nullptr, + /* update_num_ops_stats */ false); if (!merge_status.ok()) { // Failed to merge! diff --git a/table/get_context.cc b/table/get_context.cc index 41825d89e..b2daa1789 100644 --- a/table/get_context.cc +++ b/table/get_context.cc @@ -471,7 +471,8 @@ void GetContext::Merge(const Slice* value) { const Status s = MergeHelper::TimedFullMerge( merge_operator_, user_key_, value, merge_context_->GetOperands(), pinnable_val_ ? pinnable_val_->GetSelf() : nullptr, columns_, logger_, - statistics_, clock_); + statistics_, clock_, /* result_operand */ nullptr, + /* update_num_ops_stats */ true); if (!s.ok()) { state_ = kCorrupt; return; @@ -489,7 +490,7 @@ void GetContext::MergeWithEntity(Slice entity) { const Status s = MergeHelper::TimedFullMergeWithEntity( merge_operator_, user_key_, entity, merge_context_->GetOperands(), pinnable_val_ ? pinnable_val_->GetSelf() : nullptr, columns_, logger_, - statistics_, clock_); + statistics_, clock_, /* update_num_ops_stats */ true); if (!s.ok()) { state_ = kCorrupt; return; diff --git a/utilities/write_batch_with_index/write_batch_with_index_internal.cc b/utilities/write_batch_with_index/write_batch_with_index_internal.cc index 7ff6fbfaf..3c9205bf7 100644 --- a/utilities/write_batch_with_index/write_batch_with_index_internal.cc +++ b/utilities/write_batch_with_index/write_batch_with_index_internal.cc @@ -664,22 +664,25 @@ Status WriteBatchWithIndexInternal::MergeKey(const Slice& key, Statistics* statistics = immutable_db_options.statistics.get(); Logger* logger = immutable_db_options.info_log.get(); SystemClock* clock = immutable_db_options.clock; - return MergeHelper::TimedFullMerge(merge_operator, key, value, - context.GetOperands(), result, logger, - statistics, clock); + return MergeHelper::TimedFullMerge( + merge_operator, key, value, context.GetOperands(), result, logger, + statistics, clock, /* result_operand */ nullptr, + /* update_num_ops_stats */ false); } else if (db_options_ != nullptr) { Statistics* statistics = db_options_->statistics.get(); Env* env = db_options_->env; Logger* logger = db_options_->info_log.get(); SystemClock* clock = env->GetSystemClock().get(); - return MergeHelper::TimedFullMerge(merge_operator, key, value, - context.GetOperands(), result, logger, - statistics, clock); + return MergeHelper::TimedFullMerge( + merge_operator, key, value, context.GetOperands(), result, logger, + statistics, clock, /* result_operand */ nullptr, + /* update_num_ops_stats */ false); } else { const auto cf_opts = cfh->cfd()->ioptions(); return MergeHelper::TimedFullMerge( merge_operator, key, value, context.GetOperands(), result, - cf_opts->logger, cf_opts->stats, cf_opts->clock); + cf_opts->logger, cf_opts->stats, cf_opts->clock, + /* result_operand */ nullptr, /* update_num_ops_stats */ false); } } else { return Status::InvalidArgument("Must provide a column_family");