http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/48867732/thirdparty/rocksdb/db/db_iter.cc ---------------------------------------------------------------------- diff --git a/thirdparty/rocksdb/db/db_iter.cc b/thirdparty/rocksdb/db/db_iter.cc new file mode 100644 index 0000000..801b110 --- /dev/null +++ b/thirdparty/rocksdb/db/db_iter.cc @@ -0,0 +1,1256 @@ +// 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). +// +// Copyright (c) 2011 The LevelDB Authors. All rights reserved. +// Use of this source code is governed by a BSD-style license that can be +// found in the LICENSE file. See the AUTHORS file for names of contributors. + +#include "db/db_iter.h" +#include <stdexcept> +#include <deque> +#include <string> +#include <limits> + +#include "db/dbformat.h" +#include "db/merge_context.h" +#include "db/merge_helper.h" +#include "db/pinned_iterators_manager.h" +#include "monitoring/perf_context_imp.h" +#include "port/port.h" +#include "rocksdb/env.h" +#include "rocksdb/iterator.h" +#include "rocksdb/merge_operator.h" +#include "rocksdb/options.h" +#include "table/internal_iterator.h" +#include "util/arena.h" +#include "util/filename.h" +#include "util/logging.h" +#include "util/mutexlock.h" +#include "util/string_util.h" + +namespace rocksdb { + +#if 0 +static void DumpInternalIter(Iterator* iter) { + for (iter->SeekToFirst(); iter->Valid(); iter->Next()) { + ParsedInternalKey k; + if (!ParseInternalKey(iter->key(), &k)) { + fprintf(stderr, "Corrupt '%s'\n", EscapeString(iter->key()).c_str()); + } else { + fprintf(stderr, "@ '%s'\n", k.DebugString().c_str()); + } + } +} +#endif + +// Memtables and sstables that make the DB representation contain +// (userkey,seq,type) => uservalue entries. DBIter +// combines multiple entries for the same userkey found in the DB +// representation into a single entry while accounting for sequence +// numbers, deletion markers, overwrites, etc. +class DBIter: public Iterator { + public: + // The following is grossly complicated. TODO: clean it up + // Which direction is the iterator currently moving? + // (1) When moving forward, the internal iterator is positioned at + // the exact entry that yields this->key(), this->value() + // (2) When moving backwards, the internal iterator is positioned + // just before all entries whose user key == this->key(). + enum Direction { + kForward, + kReverse + }; + + // LocalStatistics contain Statistics counters that will be aggregated per + // each iterator instance and then will be sent to the global statistics when + // the iterator is destroyed. + // + // The purpose of this approach is to avoid perf regression happening + // when multiple threads bump the atomic counters from a DBIter::Next(). + struct LocalStatistics { + explicit LocalStatistics() { ResetCounters(); } + + void ResetCounters() { + next_count_ = 0; + next_found_count_ = 0; + prev_count_ = 0; + prev_found_count_ = 0; + bytes_read_ = 0; + } + + void BumpGlobalStatistics(Statistics* global_statistics) { + RecordTick(global_statistics, NUMBER_DB_NEXT, next_count_); + RecordTick(global_statistics, NUMBER_DB_NEXT_FOUND, next_found_count_); + RecordTick(global_statistics, NUMBER_DB_PREV, prev_count_); + RecordTick(global_statistics, NUMBER_DB_PREV_FOUND, prev_found_count_); + RecordTick(global_statistics, ITER_BYTES_READ, bytes_read_); + ResetCounters(); + } + + // Map to Tickers::NUMBER_DB_NEXT + uint64_t next_count_; + // Map to Tickers::NUMBER_DB_NEXT_FOUND + uint64_t next_found_count_; + // Map to Tickers::NUMBER_DB_PREV + uint64_t prev_count_; + // Map to Tickers::NUMBER_DB_PREV_FOUND + uint64_t prev_found_count_; + // Map to Tickers::ITER_BYTES_READ + uint64_t bytes_read_; + }; + + DBIter(Env* _env, const ReadOptions& read_options, + const ImmutableCFOptions& cf_options, const Comparator* cmp, + InternalIterator* iter, SequenceNumber s, bool arena_mode, + uint64_t max_sequential_skip_in_iterations) + : arena_mode_(arena_mode), + env_(_env), + logger_(cf_options.info_log), + user_comparator_(cmp), + merge_operator_(cf_options.merge_operator), + iter_(iter), + sequence_(s), + direction_(kForward), + valid_(false), + current_entry_is_merged_(false), + statistics_(cf_options.statistics), + iterate_upper_bound_(read_options.iterate_upper_bound), + prefix_same_as_start_(read_options.prefix_same_as_start), + pin_thru_lifetime_(read_options.pin_data), + total_order_seek_(read_options.total_order_seek), + range_del_agg_(cf_options.internal_comparator, s, + true /* collapse_deletions */) { + RecordTick(statistics_, NO_ITERATORS); + prefix_extractor_ = cf_options.prefix_extractor; + max_skip_ = max_sequential_skip_in_iterations; + max_skippable_internal_keys_ = read_options.max_skippable_internal_keys; + if (pin_thru_lifetime_) { + pinned_iters_mgr_.StartPinning(); + } + if (iter_) { + iter_->SetPinnedItersMgr(&pinned_iters_mgr_); + } + } + virtual ~DBIter() { + // Release pinned data if any + if (pinned_iters_mgr_.PinningEnabled()) { + pinned_iters_mgr_.ReleasePinnedData(); + } + RecordTick(statistics_, NO_ITERATORS, -1); + local_stats_.BumpGlobalStatistics(statistics_); + if (!arena_mode_) { + delete iter_; + } else { + iter_->~InternalIterator(); + } + } + virtual void SetIter(InternalIterator* iter) { + assert(iter_ == nullptr); + iter_ = iter; + iter_->SetPinnedItersMgr(&pinned_iters_mgr_); + } + virtual RangeDelAggregator* GetRangeDelAggregator() { + return &range_del_agg_; + } + + virtual bool Valid() const override { return valid_; } + virtual Slice key() const override { + assert(valid_); + return saved_key_.GetUserKey(); + } + virtual Slice value() const override { + assert(valid_); + if (current_entry_is_merged_) { + // If pinned_value_ is set then the result of merge operator is one of + // the merge operands and we should return it. + return pinned_value_.data() ? pinned_value_ : saved_value_; + } else if (direction_ == kReverse) { + return pinned_value_; + } else { + return iter_->value(); + } + } + virtual Status status() const override { + if (status_.ok()) { + return iter_->status(); + } else { + return status_; + } + } + + virtual Status GetProperty(std::string prop_name, + std::string* prop) override { + if (prop == nullptr) { + return Status::InvalidArgument("prop is nullptr"); + } + if (prop_name == "rocksdb.iterator.super-version-number") { + // First try to pass the value returned from inner iterator. + return iter_->GetProperty(prop_name, prop); + } else if (prop_name == "rocksdb.iterator.is-key-pinned") { + if (valid_) { + *prop = (pin_thru_lifetime_ && saved_key_.IsKeyPinned()) ? "1" : "0"; + } else { + *prop = "Iterator is not valid."; + } + return Status::OK(); + } + return Status::InvalidArgument("Undentified property."); + } + + virtual void Next() override; + virtual void Prev() override; + virtual void Seek(const Slice& target) override; + virtual void SeekForPrev(const Slice& target) override; + virtual void SeekToFirst() override; + virtual void SeekToLast() override; + Env* env() { return env_; } + void set_sequence(uint64_t s) { sequence_ = s; } + void set_valid(bool v) { valid_ = v; } + + private: + void ReverseToForward(); + void ReverseToBackward(); + void PrevInternal(); + void FindParseableKey(ParsedInternalKey* ikey, Direction direction); + bool FindValueForCurrentKey(); + bool FindValueForCurrentKeyUsingSeek(); + void FindPrevUserKey(); + void FindNextUserKey(); + inline void FindNextUserEntry(bool skipping, bool prefix_check); + void FindNextUserEntryInternal(bool skipping, bool prefix_check); + bool ParseKey(ParsedInternalKey* key); + void MergeValuesNewToOld(); + bool TooManyInternalKeysSkipped(bool increment = true); + + // Temporarily pin the blocks that we encounter until ReleaseTempPinnedData() + // is called + void TempPinData() { + if (!pin_thru_lifetime_) { + pinned_iters_mgr_.StartPinning(); + } + } + + // Release blocks pinned by TempPinData() + void ReleaseTempPinnedData() { + if (!pin_thru_lifetime_ && pinned_iters_mgr_.PinningEnabled()) { + pinned_iters_mgr_.ReleasePinnedData(); + } + } + + inline void ClearSavedValue() { + if (saved_value_.capacity() > 1048576) { + std::string empty; + swap(empty, saved_value_); + } else { + saved_value_.clear(); + } + } + + inline void ResetInternalKeysSkippedCounter() { + num_internal_keys_skipped_ = 0; + } + + const SliceTransform* prefix_extractor_; + bool arena_mode_; + Env* const env_; + Logger* logger_; + const Comparator* const user_comparator_; + const MergeOperator* const merge_operator_; + InternalIterator* iter_; + SequenceNumber sequence_; + + Status status_; + IterKey saved_key_; + std::string saved_value_; + Slice pinned_value_; + Direction direction_; + bool valid_; + bool current_entry_is_merged_; + // for prefix seek mode to support prev() + Statistics* statistics_; + uint64_t max_skip_; + uint64_t max_skippable_internal_keys_; + uint64_t num_internal_keys_skipped_; + const Slice* iterate_upper_bound_; + IterKey prefix_start_buf_; + Slice prefix_start_key_; + const bool prefix_same_as_start_; + // Means that we will pin all data blocks we read as long the Iterator + // is not deleted, will be true if ReadOptions::pin_data is true + const bool pin_thru_lifetime_; + const bool total_order_seek_; + // List of operands for merge operator. + MergeContext merge_context_; + RangeDelAggregator range_del_agg_; + LocalStatistics local_stats_; + PinnedIteratorsManager pinned_iters_mgr_; + + // No copying allowed + DBIter(const DBIter&); + void operator=(const DBIter&); +}; + +inline bool DBIter::ParseKey(ParsedInternalKey* ikey) { + if (!ParseInternalKey(iter_->key(), ikey)) { + status_ = Status::Corruption("corrupted internal key in DBIter"); + ROCKS_LOG_ERROR(logger_, "corrupted internal key in DBIter: %s", + iter_->key().ToString(true).c_str()); + return false; + } else { + return true; + } +} + +void DBIter::Next() { + assert(valid_); + + // Release temporarily pinned blocks from last operation + ReleaseTempPinnedData(); + ResetInternalKeysSkippedCounter(); + if (direction_ == kReverse) { + ReverseToForward(); + } else if (iter_->Valid() && !current_entry_is_merged_) { + // If the current value is not a merge, the iter position is the + // current key, which is already returned. We can safely issue a + // Next() without checking the current key. + // If the current key is a merge, very likely iter already points + // to the next internal position. + iter_->Next(); + PERF_COUNTER_ADD(internal_key_skipped_count, 1); + } + + if (statistics_ != nullptr) { + local_stats_.next_count_++; + } + // Now we point to the next internal position, for both of merge and + // not merge cases. + if (!iter_->Valid()) { + valid_ = false; + return; + } + FindNextUserEntry(true /* skipping the current user key */, prefix_same_as_start_); + if (statistics_ != nullptr && valid_) { + local_stats_.next_found_count_++; + local_stats_.bytes_read_ += (key().size() + value().size()); + } +} + +// PRE: saved_key_ has the current user key if skipping +// POST: saved_key_ should have the next user key if valid_, +// if the current entry is a result of merge +// current_entry_is_merged_ => true +// saved_value_ => the merged value +// +// NOTE: In between, saved_key_ can point to a user key that has +// a delete marker or a sequence number higher than sequence_ +// saved_key_ MUST have a proper user_key before calling this function +// +// The prefix_check parameter controls whether we check the iterated +// keys against the prefix of the seeked key. Set to false when +// performing a seek without a key (e.g. SeekToFirst). Set to +// prefix_same_as_start_ for other iterations. +inline void DBIter::FindNextUserEntry(bool skipping, bool prefix_check) { + PERF_TIMER_GUARD(find_next_user_entry_time); + FindNextUserEntryInternal(skipping, prefix_check); +} + +// Actual implementation of DBIter::FindNextUserEntry() +void DBIter::FindNextUserEntryInternal(bool skipping, bool prefix_check) { + // Loop until we hit an acceptable entry to yield + assert(iter_->Valid()); + assert(direction_ == kForward); + current_entry_is_merged_ = false; + + // How many times in a row we have skipped an entry with user key less than + // or equal to saved_key_. We could skip these entries either because + // sequence numbers were too high or because skipping = true. + // What saved_key_ contains throughout this method: + // - if skipping : saved_key_ contains the key that we need to skip, + // and we haven't seen any keys greater than that, + // - if num_skipped > 0 : saved_key_ contains the key that we have skipped + // num_skipped times, and we haven't seen any keys + // greater than that, + // - none of the above : saved_key_ can contain anything, it doesn't matter. + uint64_t num_skipped = 0; + + do { + ParsedInternalKey ikey; + + if (!ParseKey(&ikey)) { + // Skip corrupted keys. + iter_->Next(); + continue; + } + + if (iterate_upper_bound_ != nullptr && + user_comparator_->Compare(ikey.user_key, *iterate_upper_bound_) >= 0) { + break; + } + + if (prefix_extractor_ && prefix_check && + prefix_extractor_->Transform(ikey.user_key) + .compare(prefix_start_key_) != 0) { + break; + } + + if (TooManyInternalKeysSkipped()) { + return; + } + + if (ikey.sequence <= sequence_) { + if (skipping && + user_comparator_->Compare(ikey.user_key, saved_key_.GetUserKey()) <= + 0) { + num_skipped++; // skip this entry + PERF_COUNTER_ADD(internal_key_skipped_count, 1); + } else { + num_skipped = 0; + switch (ikey.type) { + case kTypeDeletion: + case kTypeSingleDeletion: + // Arrange to skip all upcoming entries for this key since + // they are hidden by this deletion. + saved_key_.SetUserKey( + ikey.user_key, + !iter_->IsKeyPinned() || !pin_thru_lifetime_ /* copy */); + skipping = true; + PERF_COUNTER_ADD(internal_delete_skipped_count, 1); + break; + case kTypeValue: + saved_key_.SetUserKey( + ikey.user_key, + !iter_->IsKeyPinned() || !pin_thru_lifetime_ /* copy */); + if (range_del_agg_.ShouldDelete( + ikey, RangeDelAggregator::RangePositioningMode:: + kForwardTraversal)) { + // Arrange to skip all upcoming entries for this key since + // they are hidden by this deletion. + skipping = true; + num_skipped = 0; + PERF_COUNTER_ADD(internal_delete_skipped_count, 1); + } else { + valid_ = true; + return; + } + break; + case kTypeMerge: + saved_key_.SetUserKey( + ikey.user_key, + !iter_->IsKeyPinned() || !pin_thru_lifetime_ /* copy */); + if (range_del_agg_.ShouldDelete( + ikey, RangeDelAggregator::RangePositioningMode:: + kForwardTraversal)) { + // Arrange to skip all upcoming entries for this key since + // they are hidden by this deletion. + skipping = true; + num_skipped = 0; + PERF_COUNTER_ADD(internal_delete_skipped_count, 1); + } else { + // By now, we are sure the current ikey is going to yield a + // value + current_entry_is_merged_ = true; + valid_ = true; + MergeValuesNewToOld(); // Go to a different state machine + return; + } + break; + default: + assert(false); + break; + } + } + } else { + // This key was inserted after our snapshot was taken. + PERF_COUNTER_ADD(internal_recent_skipped_count, 1); + + // Here saved_key_ may contain some old key, or the default empty key, or + // key assigned by some random other method. We don't care. + if (user_comparator_->Compare(ikey.user_key, saved_key_.GetUserKey()) <= + 0) { + num_skipped++; + } else { + saved_key_.SetUserKey( + ikey.user_key, + !iter_->IsKeyPinned() || !pin_thru_lifetime_ /* copy */); + skipping = false; + num_skipped = 0; + } + } + + // If we have sequentially iterated via numerous equal keys, then it's + // better to seek so that we can avoid too many key comparisons. + if (num_skipped > max_skip_) { + num_skipped = 0; + std::string last_key; + if (skipping) { + // We're looking for the next user-key but all we see are the same + // user-key with decreasing sequence numbers. Fast forward to + // sequence number 0 and type deletion (the smallest type). + AppendInternalKey(&last_key, ParsedInternalKey(saved_key_.GetUserKey(), + 0, kTypeDeletion)); + // Don't set skipping = false because we may still see more user-keys + // equal to saved_key_. + } else { + // We saw multiple entries with this user key and sequence numbers + // higher than sequence_. Fast forward to sequence_. + // Note that this only covers a case when a higher key was overwritten + // many times since our snapshot was taken, not the case when a lot of + // different keys were inserted after our snapshot was taken. + AppendInternalKey(&last_key, + ParsedInternalKey(saved_key_.GetUserKey(), sequence_, + kValueTypeForSeek)); + } + iter_->Seek(last_key); + RecordTick(statistics_, NUMBER_OF_RESEEKS_IN_ITERATION); + } else { + iter_->Next(); + } + } while (iter_->Valid()); + valid_ = false; +} + +// Merge values of the same user key starting from the current iter_ position +// Scan from the newer entries to older entries. +// PRE: iter_->key() points to the first merge type entry +// saved_key_ stores the user key +// POST: saved_value_ has the merged value for the user key +// iter_ points to the next entry (or invalid) +void DBIter::MergeValuesNewToOld() { + if (!merge_operator_) { + ROCKS_LOG_ERROR(logger_, "Options::merge_operator is null."); + status_ = Status::InvalidArgument("merge_operator_ must be set."); + valid_ = false; + return; + } + + // Temporarily pin the blocks that hold merge operands + TempPinData(); + merge_context_.Clear(); + // Start the merge process by pushing the first operand + merge_context_.PushOperand(iter_->value(), + iter_->IsValuePinned() /* operand_pinned */); + + ParsedInternalKey ikey; + Status s; + for (iter_->Next(); iter_->Valid(); iter_->Next()) { + if (!ParseKey(&ikey)) { + // skip corrupted key + continue; + } + + if (!user_comparator_->Equal(ikey.user_key, saved_key_.GetUserKey())) { + // hit the next user key, stop right here + break; + } else if (kTypeDeletion == ikey.type || kTypeSingleDeletion == ikey.type || + range_del_agg_.ShouldDelete( + ikey, RangeDelAggregator::RangePositioningMode:: + kForwardTraversal)) { + // hit a delete with the same user key, stop right here + // iter_ is positioned after delete + iter_->Next(); + break; + } else if (kTypeValue == ikey.type) { + // hit a put, merge the put value with operands and store the + // final result in saved_value_. We are done! + // ignore corruption if there is any. + const Slice val = iter_->value(); + s = MergeHelper::TimedFullMerge( + merge_operator_, ikey.user_key, &val, merge_context_.GetOperands(), + &saved_value_, logger_, statistics_, env_, &pinned_value_, true); + if (!s.ok()) { + status_ = s; + } + // iter_ is positioned after put + iter_->Next(); + return; + } else if (kTypeMerge == ikey.type) { + // hit a merge, add the value as an operand and run associative merge. + // when complete, add result to operands and continue. + merge_context_.PushOperand(iter_->value(), + iter_->IsValuePinned() /* operand_pinned */); + PERF_COUNTER_ADD(internal_merge_count, 1); + } else { + assert(false); + } + } + + // we either exhausted all internal keys under this user key, or hit + // a deletion marker. + // feed null as the existing value to the merge operator, such that + // client can differentiate this scenario and do things accordingly. + s = MergeHelper::TimedFullMerge(merge_operator_, saved_key_.GetUserKey(), + nullptr, merge_context_.GetOperands(), + &saved_value_, logger_, statistics_, env_, + &pinned_value_, true); + if (!s.ok()) { + status_ = s; + } +} + +void DBIter::Prev() { + assert(valid_); + ReleaseTempPinnedData(); + ResetInternalKeysSkippedCounter(); + if (direction_ == kForward) { + ReverseToBackward(); + } + PrevInternal(); + if (statistics_ != nullptr) { + local_stats_.prev_count_++; + if (valid_) { + local_stats_.prev_found_count_++; + local_stats_.bytes_read_ += (key().size() + value().size()); + } + } +} + +void DBIter::ReverseToForward() { + if (prefix_extractor_ != nullptr && !total_order_seek_) { + IterKey last_key; + last_key.SetInternalKey(ParsedInternalKey( + saved_key_.GetUserKey(), kMaxSequenceNumber, kValueTypeForSeek)); + iter_->Seek(last_key.GetInternalKey()); + } + FindNextUserKey(); + direction_ = kForward; + if (!iter_->Valid()) { + iter_->SeekToFirst(); + range_del_agg_.InvalidateTombstoneMapPositions(); + } +} + +void DBIter::ReverseToBackward() { + if (prefix_extractor_ != nullptr && !total_order_seek_) { + IterKey last_key; + last_key.SetInternalKey(ParsedInternalKey(saved_key_.GetUserKey(), 0, + kValueTypeForSeekForPrev)); + iter_->SeekForPrev(last_key.GetInternalKey()); + } + if (current_entry_is_merged_) { + // Not placed in the same key. Need to call Prev() until finding the + // previous key. + if (!iter_->Valid()) { + iter_->SeekToLast(); + range_del_agg_.InvalidateTombstoneMapPositions(); + } + ParsedInternalKey ikey; + FindParseableKey(&ikey, kReverse); + while (iter_->Valid() && + user_comparator_->Compare(ikey.user_key, saved_key_.GetUserKey()) > + 0) { + assert(ikey.sequence != kMaxSequenceNumber); + if (ikey.sequence > sequence_) { + PERF_COUNTER_ADD(internal_recent_skipped_count, 1); + } else { + PERF_COUNTER_ADD(internal_key_skipped_count, 1); + } + iter_->Prev(); + FindParseableKey(&ikey, kReverse); + } + } +#ifndef NDEBUG + if (iter_->Valid()) { + ParsedInternalKey ikey; + assert(ParseKey(&ikey)); + assert(user_comparator_->Compare(ikey.user_key, saved_key_.GetUserKey()) <= + 0); + } +#endif + + FindPrevUserKey(); + direction_ = kReverse; +} + +void DBIter::PrevInternal() { + if (!iter_->Valid()) { + valid_ = false; + return; + } + + ParsedInternalKey ikey; + + while (iter_->Valid()) { + saved_key_.SetUserKey( + ExtractUserKey(iter_->key()), + !iter_->IsKeyPinned() || !pin_thru_lifetime_ /* copy */); + + if (FindValueForCurrentKey()) { + valid_ = true; + if (!iter_->Valid()) { + return; + } + FindParseableKey(&ikey, kReverse); + if (user_comparator_->Equal(ikey.user_key, saved_key_.GetUserKey())) { + FindPrevUserKey(); + } + if (valid_ && prefix_extractor_ && prefix_same_as_start_ && + prefix_extractor_->Transform(saved_key_.GetUserKey()) + .compare(prefix_start_key_) != 0) { + valid_ = false; + } + return; + } + + if (TooManyInternalKeysSkipped(false)) { + return; + } + + if (!iter_->Valid()) { + break; + } + FindParseableKey(&ikey, kReverse); + if (user_comparator_->Equal(ikey.user_key, saved_key_.GetUserKey())) { + FindPrevUserKey(); + } + } + // We haven't found any key - iterator is not valid + // Or the prefix is different than start prefix + assert(!iter_->Valid()); + valid_ = false; +} + +// This function checks, if the entry with biggest sequence_number <= sequence_ +// is non kTypeDeletion or kTypeSingleDeletion. If it's not, we save value in +// saved_value_ +bool DBIter::FindValueForCurrentKey() { + assert(iter_->Valid()); + merge_context_.Clear(); + current_entry_is_merged_ = false; + // last entry before merge (could be kTypeDeletion, kTypeSingleDeletion or + // kTypeValue) + ValueType last_not_merge_type = kTypeDeletion; + ValueType last_key_entry_type = kTypeDeletion; + + ParsedInternalKey ikey; + FindParseableKey(&ikey, kReverse); + + // Temporarily pin blocks that hold (merge operands / the value) + ReleaseTempPinnedData(); + TempPinData(); + size_t num_skipped = 0; + while (iter_->Valid() && ikey.sequence <= sequence_ && + user_comparator_->Equal(ikey.user_key, saved_key_.GetUserKey())) { + if (TooManyInternalKeysSkipped()) { + return false; + } + + // We iterate too much: let's use Seek() to avoid too much key comparisons + if (num_skipped >= max_skip_) { + return FindValueForCurrentKeyUsingSeek(); + } + + last_key_entry_type = ikey.type; + switch (last_key_entry_type) { + case kTypeValue: + if (range_del_agg_.ShouldDelete( + ikey, + RangeDelAggregator::RangePositioningMode::kBackwardTraversal)) { + last_key_entry_type = kTypeRangeDeletion; + PERF_COUNTER_ADD(internal_delete_skipped_count, 1); + } else { + assert(iter_->IsValuePinned()); + pinned_value_ = iter_->value(); + } + merge_context_.Clear(); + last_not_merge_type = last_key_entry_type; + break; + case kTypeDeletion: + case kTypeSingleDeletion: + merge_context_.Clear(); + last_not_merge_type = last_key_entry_type; + PERF_COUNTER_ADD(internal_delete_skipped_count, 1); + break; + case kTypeMerge: + if (range_del_agg_.ShouldDelete( + ikey, + RangeDelAggregator::RangePositioningMode::kBackwardTraversal)) { + merge_context_.Clear(); + last_key_entry_type = kTypeRangeDeletion; + last_not_merge_type = last_key_entry_type; + PERF_COUNTER_ADD(internal_delete_skipped_count, 1); + } else { + assert(merge_operator_ != nullptr); + merge_context_.PushOperandBack( + iter_->value(), iter_->IsValuePinned() /* operand_pinned */); + PERF_COUNTER_ADD(internal_merge_count, 1); + } + break; + default: + assert(false); + } + + PERF_COUNTER_ADD(internal_key_skipped_count, 1); + assert(user_comparator_->Equal(ikey.user_key, saved_key_.GetUserKey())); + iter_->Prev(); + ++num_skipped; + FindParseableKey(&ikey, kReverse); + } + + Status s; + switch (last_key_entry_type) { + case kTypeDeletion: + case kTypeSingleDeletion: + case kTypeRangeDeletion: + valid_ = false; + return false; + case kTypeMerge: + current_entry_is_merged_ = true; + if (last_not_merge_type == kTypeDeletion || + last_not_merge_type == kTypeSingleDeletion || + last_not_merge_type == kTypeRangeDeletion) { + s = MergeHelper::TimedFullMerge( + merge_operator_, saved_key_.GetUserKey(), nullptr, + merge_context_.GetOperands(), &saved_value_, logger_, statistics_, + env_, &pinned_value_, true); + } else { + assert(last_not_merge_type == kTypeValue); + s = MergeHelper::TimedFullMerge( + merge_operator_, saved_key_.GetUserKey(), &pinned_value_, + merge_context_.GetOperands(), &saved_value_, logger_, statistics_, + env_, &pinned_value_, true); + } + break; + case kTypeValue: + // do nothing - we've already has value in saved_value_ + break; + default: + assert(false); + break; + } + valid_ = true; + if (!s.ok()) { + status_ = s; + } + return true; +} + +// This function is used in FindValueForCurrentKey. +// We use Seek() function instead of Prev() to find necessary value +bool DBIter::FindValueForCurrentKeyUsingSeek() { + // FindValueForCurrentKey will enable pinning before calling + // FindValueForCurrentKeyUsingSeek() + assert(pinned_iters_mgr_.PinningEnabled()); + std::string last_key; + AppendInternalKey(&last_key, ParsedInternalKey(saved_key_.GetUserKey(), + sequence_, kValueTypeForSeek)); + iter_->Seek(last_key); + RecordTick(statistics_, NUMBER_OF_RESEEKS_IN_ITERATION); + + // assume there is at least one parseable key for this user key + ParsedInternalKey ikey; + FindParseableKey(&ikey, kForward); + + if (ikey.type == kTypeDeletion || ikey.type == kTypeSingleDeletion || + range_del_agg_.ShouldDelete( + ikey, RangeDelAggregator::RangePositioningMode::kBackwardTraversal)) { + valid_ = false; + return false; + } + if (ikey.type == kTypeValue) { + assert(iter_->IsValuePinned()); + pinned_value_ = iter_->value(); + valid_ = true; + return true; + } + + // kTypeMerge. We need to collect all kTypeMerge values and save them + // in operands + current_entry_is_merged_ = true; + merge_context_.Clear(); + while ( + iter_->Valid() && + user_comparator_->Equal(ikey.user_key, saved_key_.GetUserKey()) && + ikey.type == kTypeMerge && + !range_del_agg_.ShouldDelete( + ikey, RangeDelAggregator::RangePositioningMode::kBackwardTraversal)) { + merge_context_.PushOperand(iter_->value(), + iter_->IsValuePinned() /* operand_pinned */); + PERF_COUNTER_ADD(internal_merge_count, 1); + iter_->Next(); + FindParseableKey(&ikey, kForward); + } + + Status s; + if (!iter_->Valid() || + !user_comparator_->Equal(ikey.user_key, saved_key_.GetUserKey()) || + ikey.type == kTypeDeletion || ikey.type == kTypeSingleDeletion || + range_del_agg_.ShouldDelete( + ikey, RangeDelAggregator::RangePositioningMode::kBackwardTraversal)) { + s = MergeHelper::TimedFullMerge(merge_operator_, saved_key_.GetUserKey(), + nullptr, merge_context_.GetOperands(), + &saved_value_, logger_, statistics_, env_, + &pinned_value_, true); + // Make iter_ valid and point to saved_key_ + if (!iter_->Valid() || + !user_comparator_->Equal(ikey.user_key, saved_key_.GetUserKey())) { + iter_->Seek(last_key); + RecordTick(statistics_, NUMBER_OF_RESEEKS_IN_ITERATION); + } + valid_ = true; + if (!s.ok()) { + status_ = s; + } + return true; + } + + const Slice& val = iter_->value(); + s = MergeHelper::TimedFullMerge(merge_operator_, saved_key_.GetUserKey(), + &val, merge_context_.GetOperands(), + &saved_value_, logger_, statistics_, env_, + &pinned_value_, true); + valid_ = true; + if (!s.ok()) { + status_ = s; + } + return true; +} + +// Used in Next to change directions +// Go to next user key +// Don't use Seek(), +// because next user key will be very close +void DBIter::FindNextUserKey() { + if (!iter_->Valid()) { + return; + } + ParsedInternalKey ikey; + FindParseableKey(&ikey, kForward); + while (iter_->Valid() && + !user_comparator_->Equal(ikey.user_key, saved_key_.GetUserKey())) { + iter_->Next(); + FindParseableKey(&ikey, kForward); + } +} + +// Go to previous user_key +void DBIter::FindPrevUserKey() { + if (!iter_->Valid()) { + return; + } + size_t num_skipped = 0; + ParsedInternalKey ikey; + FindParseableKey(&ikey, kReverse); + int cmp; + while (iter_->Valid() && + ((cmp = user_comparator_->Compare(ikey.user_key, + saved_key_.GetUserKey())) == 0 || + (cmp > 0 && ikey.sequence > sequence_))) { + if (TooManyInternalKeysSkipped()) { + return; + } + + if (cmp == 0) { + if (num_skipped >= max_skip_) { + num_skipped = 0; + IterKey last_key; + last_key.SetInternalKey(ParsedInternalKey( + saved_key_.GetUserKey(), kMaxSequenceNumber, kValueTypeForSeek)); + iter_->Seek(last_key.GetInternalKey()); + RecordTick(statistics_, NUMBER_OF_RESEEKS_IN_ITERATION); + } else { + ++num_skipped; + } + } + assert(ikey.sequence != kMaxSequenceNumber); + if (ikey.sequence > sequence_) { + PERF_COUNTER_ADD(internal_recent_skipped_count, 1); + } else { + PERF_COUNTER_ADD(internal_key_skipped_count, 1); + } + iter_->Prev(); + FindParseableKey(&ikey, kReverse); + } +} + +bool DBIter::TooManyInternalKeysSkipped(bool increment) { + if ((max_skippable_internal_keys_ > 0) && + (num_internal_keys_skipped_ > max_skippable_internal_keys_)) { + valid_ = false; + status_ = Status::Incomplete("Too many internal keys skipped."); + return true; + } else if (increment) { + num_internal_keys_skipped_++; + } + return false; +} + +// Skip all unparseable keys +void DBIter::FindParseableKey(ParsedInternalKey* ikey, Direction direction) { + while (iter_->Valid() && !ParseKey(ikey)) { + if (direction == kReverse) { + iter_->Prev(); + } else { + iter_->Next(); + } + } +} + +void DBIter::Seek(const Slice& target) { + StopWatch sw(env_, statistics_, DB_SEEK); + ReleaseTempPinnedData(); + ResetInternalKeysSkippedCounter(); + saved_key_.Clear(); + saved_key_.SetInternalKey(target, sequence_); + + { + PERF_TIMER_GUARD(seek_internal_seek_time); + iter_->Seek(saved_key_.GetInternalKey()); + range_del_agg_.InvalidateTombstoneMapPositions(); + } + RecordTick(statistics_, NUMBER_DB_SEEK); + if (iter_->Valid()) { + if (prefix_extractor_ && prefix_same_as_start_) { + prefix_start_key_ = prefix_extractor_->Transform(target); + } + direction_ = kForward; + ClearSavedValue(); + FindNextUserEntry(false /* not skipping */, prefix_same_as_start_); + if (!valid_) { + prefix_start_key_.clear(); + } + if (statistics_ != nullptr) { + if (valid_) { + RecordTick(statistics_, NUMBER_DB_SEEK_FOUND); + RecordTick(statistics_, ITER_BYTES_READ, key().size() + value().size()); + } + } + } else { + valid_ = false; + } + + if (valid_ && prefix_extractor_ && prefix_same_as_start_) { + prefix_start_buf_.SetUserKey(prefix_start_key_); + prefix_start_key_ = prefix_start_buf_.GetUserKey(); + } +} + +void DBIter::SeekForPrev(const Slice& target) { + StopWatch sw(env_, statistics_, DB_SEEK); + ReleaseTempPinnedData(); + ResetInternalKeysSkippedCounter(); + saved_key_.Clear(); + // now saved_key is used to store internal key. + saved_key_.SetInternalKey(target, 0 /* sequence_number */, + kValueTypeForSeekForPrev); + + { + PERF_TIMER_GUARD(seek_internal_seek_time); + iter_->SeekForPrev(saved_key_.GetInternalKey()); + range_del_agg_.InvalidateTombstoneMapPositions(); + } + + RecordTick(statistics_, NUMBER_DB_SEEK); + if (iter_->Valid()) { + if (prefix_extractor_ && prefix_same_as_start_) { + prefix_start_key_ = prefix_extractor_->Transform(target); + } + direction_ = kReverse; + ClearSavedValue(); + PrevInternal(); + if (!valid_) { + prefix_start_key_.clear(); + } + if (statistics_ != nullptr) { + if (valid_) { + RecordTick(statistics_, NUMBER_DB_SEEK_FOUND); + RecordTick(statistics_, ITER_BYTES_READ, key().size() + value().size()); + } + } + } else { + valid_ = false; + } + if (valid_ && prefix_extractor_ && prefix_same_as_start_) { + prefix_start_buf_.SetUserKey(prefix_start_key_); + prefix_start_key_ = prefix_start_buf_.GetUserKey(); + } +} + +void DBIter::SeekToFirst() { + // Don't use iter_::Seek() if we set a prefix extractor + // because prefix seek will be used. + if (prefix_extractor_ != nullptr) { + max_skip_ = std::numeric_limits<uint64_t>::max(); + } + direction_ = kForward; + ReleaseTempPinnedData(); + ResetInternalKeysSkippedCounter(); + ClearSavedValue(); + + { + PERF_TIMER_GUARD(seek_internal_seek_time); + iter_->SeekToFirst(); + range_del_agg_.InvalidateTombstoneMapPositions(); + } + + RecordTick(statistics_, NUMBER_DB_SEEK); + if (iter_->Valid()) { + saved_key_.SetUserKey( + ExtractUserKey(iter_->key()), + !iter_->IsKeyPinned() || !pin_thru_lifetime_ /* copy */); + FindNextUserEntry(false /* not skipping */, false /* no prefix check */); + if (statistics_ != nullptr) { + if (valid_) { + RecordTick(statistics_, NUMBER_DB_SEEK_FOUND); + RecordTick(statistics_, ITER_BYTES_READ, key().size() + value().size()); + } + } + } else { + valid_ = false; + } + if (valid_ && prefix_extractor_ && prefix_same_as_start_) { + prefix_start_buf_.SetUserKey( + prefix_extractor_->Transform(saved_key_.GetUserKey())); + prefix_start_key_ = prefix_start_buf_.GetUserKey(); + } +} + +void DBIter::SeekToLast() { + // Don't use iter_::Seek() if we set a prefix extractor + // because prefix seek will be used. + if (prefix_extractor_ != nullptr) { + max_skip_ = std::numeric_limits<uint64_t>::max(); + } + direction_ = kReverse; + ReleaseTempPinnedData(); + ResetInternalKeysSkippedCounter(); + ClearSavedValue(); + + { + PERF_TIMER_GUARD(seek_internal_seek_time); + iter_->SeekToLast(); + range_del_agg_.InvalidateTombstoneMapPositions(); + } + // When the iterate_upper_bound is set to a value, + // it will seek to the last key before the + // ReadOptions.iterate_upper_bound + if (iter_->Valid() && iterate_upper_bound_ != nullptr) { + SeekForPrev(*iterate_upper_bound_); + range_del_agg_.InvalidateTombstoneMapPositions(); + if (!Valid()) { + return; + } else if (user_comparator_->Equal(*iterate_upper_bound_, key())) { + Prev(); + } + } else { + PrevInternal(); + } + if (statistics_ != nullptr) { + RecordTick(statistics_, NUMBER_DB_SEEK); + if (valid_) { + RecordTick(statistics_, NUMBER_DB_SEEK_FOUND); + RecordTick(statistics_, ITER_BYTES_READ, key().size() + value().size()); + } + } + if (valid_ && prefix_extractor_ && prefix_same_as_start_) { + prefix_start_buf_.SetUserKey( + prefix_extractor_->Transform(saved_key_.GetUserKey())); + prefix_start_key_ = prefix_start_buf_.GetUserKey(); + } +} + +Iterator* NewDBIterator(Env* env, const ReadOptions& read_options, + const ImmutableCFOptions& cf_options, + const Comparator* user_key_comparator, + InternalIterator* internal_iter, + const SequenceNumber& sequence, + uint64_t max_sequential_skip_in_iterations) { + DBIter* db_iter = new DBIter(env, read_options, cf_options, + user_key_comparator, internal_iter, sequence, + false, max_sequential_skip_in_iterations); + return db_iter; +} + +ArenaWrappedDBIter::~ArenaWrappedDBIter() { db_iter_->~DBIter(); } + +RangeDelAggregator* ArenaWrappedDBIter::GetRangeDelAggregator() { + return db_iter_->GetRangeDelAggregator(); +} + +void ArenaWrappedDBIter::SetIterUnderDBIter(InternalIterator* iter) { + static_cast<DBIter*>(db_iter_)->SetIter(iter); +} + +inline bool ArenaWrappedDBIter::Valid() const { return db_iter_->Valid(); } +inline void ArenaWrappedDBIter::SeekToFirst() { db_iter_->SeekToFirst(); } +inline void ArenaWrappedDBIter::SeekToLast() { db_iter_->SeekToLast(); } +inline void ArenaWrappedDBIter::Seek(const Slice& target) { + db_iter_->Seek(target); +} +inline void ArenaWrappedDBIter::SeekForPrev(const Slice& target) { + db_iter_->SeekForPrev(target); +} +inline void ArenaWrappedDBIter::Next() { db_iter_->Next(); } +inline void ArenaWrappedDBIter::Prev() { db_iter_->Prev(); } +inline Slice ArenaWrappedDBIter::key() const { return db_iter_->key(); } +inline Slice ArenaWrappedDBIter::value() const { return db_iter_->value(); } +inline Status ArenaWrappedDBIter::status() const { return db_iter_->status(); } +inline Status ArenaWrappedDBIter::GetProperty(std::string prop_name, + std::string* prop) { + if (prop_name == "rocksdb.iterator.super-version-number") { + // First try to pass the value returned from inner iterator. + if (!db_iter_->GetProperty(prop_name, prop).ok()) { + *prop = ToString(sv_number_); + } + return Status::OK(); + } + return db_iter_->GetProperty(prop_name, prop); +} + +void ArenaWrappedDBIter::Init(Env* env, const ReadOptions& read_options, + const ImmutableCFOptions& cf_options, + const SequenceNumber& sequence, + uint64_t max_sequential_skip_in_iteration, + uint64_t version_number) { + auto mem = arena_.AllocateAligned(sizeof(DBIter)); + db_iter_ = new (mem) + DBIter(env, read_options, cf_options, cf_options.user_comparator, nullptr, + sequence, true, max_sequential_skip_in_iteration); + sv_number_ = version_number; +} + +Status ArenaWrappedDBIter::Refresh() { + if (cfd_ == nullptr || db_impl_ == nullptr) { + return Status::NotSupported("Creating renew iterator is not allowed."); + } + assert(db_iter_ != nullptr); + SequenceNumber latest_seq = db_impl_->GetLatestSequenceNumber(); + uint64_t cur_sv_number = cfd_->GetSuperVersionNumber(); + if (sv_number_ != cur_sv_number) { + Env* env = db_iter_->env(); + db_iter_->~DBIter(); + arena_.~Arena(); + new (&arena_) Arena(); + + SuperVersion* sv = cfd_->GetReferencedSuperVersion(db_impl_->mutex()); + Init(env, read_options_, *(cfd_->ioptions()), latest_seq, + sv->mutable_cf_options.max_sequential_skip_in_iterations, + cur_sv_number); + + InternalIterator* internal_iter = db_impl_->NewInternalIterator( + read_options_, cfd_, sv, &arena_, db_iter_->GetRangeDelAggregator()); + SetIterUnderDBIter(internal_iter); + } else { + db_iter_->set_sequence(latest_seq); + db_iter_->set_valid(false); + } + return Status::OK(); +} + +ArenaWrappedDBIter* NewArenaWrappedDbIterator( + Env* env, const ReadOptions& read_options, + const ImmutableCFOptions& cf_options, const SequenceNumber& sequence, + uint64_t max_sequential_skip_in_iterations, uint64_t version_number, + DBImpl* db_impl, ColumnFamilyData* cfd) { + ArenaWrappedDBIter* iter = new ArenaWrappedDBIter(); + iter->Init(env, read_options, cf_options, sequence, + max_sequential_skip_in_iterations, version_number); + if (db_impl != nullptr && cfd != nullptr) { + iter->StoreRefreshInfo(read_options, db_impl, cfd); + } + + return iter; +} + +} // namespace rocksdb
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/48867732/thirdparty/rocksdb/db/db_iter.h ---------------------------------------------------------------------- diff --git a/thirdparty/rocksdb/db/db_iter.h b/thirdparty/rocksdb/db/db_iter.h new file mode 100644 index 0000000..ea98ff4 --- /dev/null +++ b/thirdparty/rocksdb/db/db_iter.h @@ -0,0 +1,100 @@ +// 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). +// +// Copyright (c) 2011 The LevelDB Authors. All rights reserved. +// Use of this source code is governed by a BSD-style license that can be +// found in the LICENSE file. See the AUTHORS file for names of contributors. + +#pragma once +#include <stdint.h> +#include <string> +#include "db/db_impl.h" +#include "db/dbformat.h" +#include "db/range_del_aggregator.h" +#include "options/cf_options.h" +#include "rocksdb/db.h" +#include "rocksdb/iterator.h" +#include "util/arena.h" +#include "util/autovector.h" + +namespace rocksdb { + +class Arena; +class DBIter; +class InternalIterator; + +// Return a new iterator that converts internal keys (yielded by +// "*internal_iter") that were live at the specified "sequence" number +// into appropriate user keys. +extern Iterator* NewDBIterator(Env* env, const ReadOptions& read_options, + const ImmutableCFOptions& cf_options, + const Comparator* user_key_comparator, + InternalIterator* internal_iter, + const SequenceNumber& sequence, + uint64_t max_sequential_skip_in_iterations); + +// A wrapper iterator which wraps DB Iterator and the arena, with which the DB +// iterator is supposed be allocated. This class is used as an entry point of +// a iterator hierarchy whose memory can be allocated inline. In that way, +// accessing the iterator tree can be more cache friendly. It is also faster +// to allocate. +class ArenaWrappedDBIter : public Iterator { + public: + virtual ~ArenaWrappedDBIter(); + + // Get the arena to be used to allocate memory for DBIter to be wrapped, + // as well as child iterators in it. + virtual Arena* GetArena() { return &arena_; } + virtual RangeDelAggregator* GetRangeDelAggregator(); + + // Set the internal iterator wrapped inside the DB Iterator. Usually it is + // a merging iterator. + virtual void SetIterUnderDBIter(InternalIterator* iter); + virtual bool Valid() const override; + virtual void SeekToFirst() override; + virtual void SeekToLast() override; + virtual void Seek(const Slice& target) override; + virtual void SeekForPrev(const Slice& target) override; + virtual void Next() override; + virtual void Prev() override; + virtual Slice key() const override; + virtual Slice value() const override; + virtual Status status() const override; + virtual Status Refresh() override; + + virtual Status GetProperty(std::string prop_name, std::string* prop) override; + + void Init(Env* env, const ReadOptions& read_options, + const ImmutableCFOptions& cf_options, + const SequenceNumber& sequence, + uint64_t max_sequential_skip_in_iterations, + uint64_t version_number); + + void StoreRefreshInfo(const ReadOptions& read_options, DBImpl* db_impl, + ColumnFamilyData* cfd) { + read_options_ = read_options; + db_impl_ = db_impl; + cfd_ = cfd; + } + + private: + DBIter* db_iter_; + Arena arena_; + uint64_t sv_number_; + ColumnFamilyData* cfd_ = nullptr; + DBImpl* db_impl_ = nullptr; + ReadOptions read_options_; +}; + +// Generate the arena wrapped iterator class. +// `db_impl` and `cfd` are used for reneweal. If left null, renewal will not +// be supported. +extern ArenaWrappedDBIter* NewArenaWrappedDbIterator( + Env* env, const ReadOptions& read_options, + const ImmutableCFOptions& cf_options, const SequenceNumber& sequence, + uint64_t max_sequential_skip_in_iterations, uint64_t version_number, + DBImpl* db_impl = nullptr, ColumnFamilyData* cfd = nullptr); + +} // namespace rocksdb
