db_iter.cc 55.4 KB
Newer Older
1
//  Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
S
Siying Dong 已提交
2 3 4
//  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).
5
//
J
jorlow@chromium.org 已提交
6 7 8 9 10
// 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"
S
Stanislau Hlebik 已提交
11
#include <string>
12
#include <iostream>
S
Stanislau Hlebik 已提交
13
#include <limits>
J
jorlow@chromium.org 已提交
14 15

#include "db/dbformat.h"
16
#include "db/merge_context.h"
17
#include "db/merge_helper.h"
18
#include "db/pinned_iterators_manager.h"
19
#include "file/filename.h"
20
#include "memory/arena.h"
21
#include "monitoring/perf_context_imp.h"
22 23 24
#include "rocksdb/env.h"
#include "rocksdb/iterator.h"
#include "rocksdb/merge_operator.h"
25
#include "rocksdb/options.h"
S
sdong 已提交
26
#include "table/internal_iterator.h"
27
#include "table/iterator_wrapper.h"
J
jorlow@chromium.org 已提交
28 29
#include "util/logging.h"
#include "util/mutexlock.h"
30
#include "util/string_util.h"
31
#include "util/trace_replay.h"
32
#include "util/user_comparator_wrapper.h"
J
jorlow@chromium.org 已提交
33

34
namespace rocksdb {
J
jorlow@chromium.org 已提交
35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53

#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.
S
Siying Dong 已提交
54
class DBIter final: public Iterator {
J
jorlow@chromium.org 已提交
55
 public:
56
  // The following is grossly complicated. TODO: clean it up
J
jorlow@chromium.org 已提交
57
  // Which direction is the iterator currently moving?
58 59 60 61 62 63 64
  // (1) When moving forward:
  //   (1a) if current_entry_is_merged_ = false, the internal iterator is
  //        positioned at the exact entry that yields this->key(), this->value()
  //   (1b) if current_entry_is_merged_ = true, the internal iterator is
  //        positioned immediately after the last entry that contributed to the
  //        current this->value(). That entry may or may not have key equal to
  //        this->key().
J
jorlow@chromium.org 已提交
65 66 67 68 69 70 71
  // (2) When moving backwards, the internal iterator is positioned
  //     just before all entries whose user key == this->key().
  enum Direction {
    kForward,
    kReverse
  };

72 73 74 75 76 77 78 79 80 81 82 83 84 85 86
  // 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;
87
      skip_count_ = 0;
88 89 90 91 92 93 94 95
    }

    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_);
96
      RecordTick(global_statistics, NUMBER_ITER_SKIP, skip_count_);
97
      PERF_COUNTER_ADD(iter_read_bytes, bytes_read_);
98 99 100 101 102 103 104 105 106 107 108 109 110
      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_;
111 112
    // Map to Tickers::NUMBER_ITER_SKIP
    uint64_t skip_count_;
113 114
  };

S
Siying Dong 已提交
115
  DBIter(Env* _env, const ReadOptions& read_options,
116 117
         const ImmutableCFOptions& cf_options,
         const MutableCFOptions& mutable_cf_options, const Comparator* cmp,
S
sdong 已提交
118
         InternalIterator* iter, SequenceNumber s, bool arena_mode,
Y
Yi Wu 已提交
119
         uint64_t max_sequential_skip_in_iterations,
120 121
         ReadCallback* read_callback, DBImpl* db_impl, ColumnFamilyData* cfd,
         bool allow_blob)
122
      : env_(_env),
123
        logger_(cf_options.info_log),
J
jorlow@chromium.org 已提交
124
        user_comparator_(cmp),
125
        merge_operator_(cf_options.merge_operator),
J
jorlow@chromium.org 已提交
126
        iter_(iter),
127
        read_callback_(read_callback),
J
jorlow@chromium.org 已提交
128
        sequence_(s),
129
        statistics_(cf_options.statistics),
130
        num_internal_keys_skipped_(0),
131
        iterate_lower_bound_(read_options.iterate_lower_bound),
132
        iterate_upper_bound_(read_options.iterate_upper_bound),
133 134 135
        direction_(kForward),
        valid_(false),
        current_entry_is_merged_(false),
136
        is_key_seqnum_zero_(false),
137 138 139
        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),
140 141 142
        allow_blob_(allow_blob),
        is_blob_(false),
        arena_mode_(arena_mode),
143
        range_del_agg_(&cf_options.internal_comparator, s),
144 145
        db_impl_(db_impl),
        cfd_(cfd),
146
        start_seqnum_(read_options.iter_start_seqnum) {
147
    RecordTick(statistics_, NO_ITERATOR_CREATED);
148
    prefix_extractor_ = mutable_cf_options.prefix_extractor.get();
149
    max_skip_ = max_sequential_skip_in_iterations;
150
    max_skippable_internal_keys_ = read_options.max_skippable_internal_keys;
151 152 153
    if (pin_thru_lifetime_) {
      pinned_iters_mgr_.StartPinning();
    }
154 155
    if (iter_.iter()) {
      iter_.iter()->SetPinnedItersMgr(&pinned_iters_mgr_);
156
    }
J
jorlow@chromium.org 已提交
157
  }
158
  ~DBIter() override {
159
    // Release pinned data if any
160 161 162
    if (pinned_iters_mgr_.PinningEnabled()) {
      pinned_iters_mgr_.ReleasePinnedData();
    }
163
    RecordTick(statistics_, NO_ITERATOR_DELETED);
164
    ResetInternalKeysSkippedCounter();
165
    local_stats_.BumpGlobalStatistics(statistics_);
166
    iter_.DeleteIter(arena_mode_);
167
  }
S
sdong 已提交
168
  virtual void SetIter(InternalIterator* iter) {
169 170 171
    assert(iter_.iter() == nullptr);
    iter_.Set(iter);
    iter_.iter()->SetPinnedItersMgr(&pinned_iters_mgr_);
J
jorlow@chromium.org 已提交
172
  }
173
  virtual ReadRangeDelAggregator* GetRangeDelAggregator() {
A
Andrew Kryczka 已提交
174 175 176
    return &range_del_agg_;
  }

177 178
  bool Valid() const override { return valid_; }
  Slice key() const override {
J
jorlow@chromium.org 已提交
179
    assert(valid_);
180 181 182 183 184
    if(start_seqnum_ > 0) {
      return saved_key_.GetInternalKey();
    } else {
      return saved_key_.GetUserKey();
    }
J
jorlow@chromium.org 已提交
185
  }
186
  Slice value() const override {
J
jorlow@chromium.org 已提交
187
    assert(valid_);
188
    if (current_entry_is_merged_) {
189 190 191
      // 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_;
192 193 194
    } else if (direction_ == kReverse) {
      return pinned_value_;
    } else {
195
      return iter_.value();
196
    }
J
jorlow@chromium.org 已提交
197
  }
198
  Status status() const override {
J
jorlow@chromium.org 已提交
199
    if (status_.ok()) {
200
      return iter_.status();
J
jorlow@chromium.org 已提交
201
    } else {
202
      assert(!valid_);
J
jorlow@chromium.org 已提交
203 204 205
      return status_;
    }
  }
Y
Yi Wu 已提交
206 207 208 209
  bool IsBlob() const {
    assert(valid_ && (allow_blob_ || !is_blob_));
    return is_blob_;
  }
210

211
  Status GetProperty(std::string prop_name, std::string* prop) override {
212 213 214
    if (prop == nullptr) {
      return Status::InvalidArgument("prop is nullptr");
    }
215
    if (prop_name == "rocksdb.iterator.super-version-number") {
216
      // First try to pass the value returned from inner iterator.
217
      return iter_.iter()->GetProperty(prop_name, prop);
218
    } else if (prop_name == "rocksdb.iterator.is-key-pinned") {
219
      if (valid_) {
220
        *prop = (pin_thru_lifetime_ && saved_key_.IsKeyPinned()) ? "1" : "0";
221 222 223 224
      } else {
        *prop = "Iterator is not valid.";
      }
      return Status::OK();
225 226 227
    } else if (prop_name == "rocksdb.iterator.internal-key") {
      *prop = saved_key_.GetUserKey().ToString();
      return Status::OK();
228
    }
229
    return Status::InvalidArgument("Unidentified property.");
230
  }
J
jorlow@chromium.org 已提交
231

232 233 234 235 236 237
  inline void Next() final override;
  inline void Prev() final override;
  inline void Seek(const Slice& target) final override;
  inline void SeekForPrev(const Slice& target) final override;
  inline void SeekToFirst() final override;
  inline void SeekToLast() final override;
S
Siying Dong 已提交
238
  Env* env() { return env_; }
239 240 241 242 243 244
  void set_sequence(uint64_t s) {
    sequence_ = s;
    if (read_callback_) {
      read_callback_->Refresh(s);
    }
  }
S
Siying Dong 已提交
245
  void set_valid(bool v) { valid_ = v; }
J
jorlow@chromium.org 已提交
246

J
jorlow@chromium.org 已提交
247
 private:
248 249 250 251 252 253
  // For all methods in this block:
  // PRE: iter_->Valid() && status_.ok()
  // Return false if there was an error, and status() is non-ok, valid_ = false;
  // in this case callers would usually stop what they were doing and return.
  bool ReverseToForward();
  bool ReverseToBackward();
S
Stanislau Hlebik 已提交
254 255
  bool FindValueForCurrentKey();
  bool FindValueForCurrentKeyUsingSeek();
256 257
  bool FindUserKeyBeforeSavedKey();
  inline bool FindNextUserEntry(bool skipping, bool prefix_check);
258
  inline bool FindNextUserEntryInternal(bool skipping, bool prefix_check);
J
jorlow@chromium.org 已提交
259
  bool ParseKey(ParsedInternalKey* key);
260 261 262
  bool MergeValuesNewToOld();

  void PrevInternal();
263
  bool TooManyInternalKeysSkipped(bool increment = true);
264
  inline bool IsVisible(SequenceNumber sequence);
J
jorlow@chromium.org 已提交
265

266 267 268 269 270 271
  // CanReseekToSkip() returns whether the iterator can use the optimization
  // where it reseek by sequence number to get the next key when there are too
  // many versions. This is disabled for write unprepared because seeking to
  // sequence number does not guarantee that it is visible.
  inline bool CanReseekToSkip();

272 273 274 275 276 277 278 279 280 281
  // 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() {
282 283
    if (!pin_thru_lifetime_ && pinned_iters_mgr_.PinningEnabled()) {
      pinned_iters_mgr_.ReleasePinnedData();
284 285 286
    }
  }

J
jorlow@chromium.org 已提交
287 288 289 290 291 292 293 294 295
  inline void ClearSavedValue() {
    if (saved_value_.capacity() > 1048576) {
      std::string empty;
      swap(empty, saved_value_);
    } else {
      saved_value_.clear();
    }
  }

296
  inline void ResetInternalKeysSkippedCounter() {
297 298 299 300
    local_stats_.skip_count_ += num_internal_keys_skipped_;
    if (valid_) {
      local_stats_.skip_count_--;
    }
301 302 303
    num_internal_keys_skipped_ = 0;
  }

304
  const SliceTransform* prefix_extractor_;
J
jorlow@chromium.org 已提交
305
  Env* const env_;
I
Igor Canadi 已提交
306
  Logger* logger_;
307
  UserComparatorWrapper user_comparator_;
308
  const MergeOperator* const merge_operator_;
309
  IteratorWrapper iter_;
310
  ReadCallback* read_callback_;
311 312
  // Max visible sequence number. It is normally the snapshot seq unless we have
  // uncommitted data in db as in WriteUnCommitted.
S
Siying Dong 已提交
313
  SequenceNumber sequence_;
J
jorlow@chromium.org 已提交
314

S
Stanislau Hlebik 已提交
315
  IterKey saved_key_;
316 317 318 319
  // Reusable internal key data structure. This is only used inside one function
  // and should not be used across functions. Reusing this object can reduce
  // overhead of calling construction of the function if creating it each time.
  ParsedInternalKey ikey_;
S
Stanislau Hlebik 已提交
320
  std::string saved_value_;
321
  Slice pinned_value_;
322
  // for prefix seek mode to support prev()
323
  Statistics* statistics_;
324
  uint64_t max_skip_;
325 326
  uint64_t max_skippable_internal_keys_;
  uint64_t num_internal_keys_skipped_;
327
  const Slice* iterate_lower_bound_;
328
  const Slice* iterate_upper_bound_;
329

330
  IterKey prefix_start_buf_;
331 332

  Status status_;
333
  Slice prefix_start_key_;
334 335 336
  Direction direction_;
  bool valid_;
  bool current_entry_is_merged_;
337 338 339 340
  // True if we know that the current entry's seqnum is 0.
  // This information is used as that the next entry will be for another
  // user key.
  bool is_key_seqnum_zero_;
341
  const bool prefix_same_as_start_;
342 343 344
  // 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_;
345
  const bool total_order_seek_;
346 347 348
  bool allow_blob_;
  bool is_blob_;
  bool arena_mode_;
349
  // List of operands for merge operator.
350
  MergeContext merge_context_;
351
  ReadRangeDelAggregator range_del_agg_;
352
  LocalStatistics local_stats_;
353
  PinnedIteratorsManager pinned_iters_mgr_;
354 355
  DBImpl* db_impl_;
  ColumnFamilyData* cfd_;
356 357 358
  // for diff snapshots we want the lower bound on the seqnum;
  // if this value > 0 iterator will return internal keys
  SequenceNumber start_seqnum_;
J
jorlow@chromium.org 已提交
359 360 361 362 363 364 365

  // No copying allowed
  DBIter(const DBIter&);
  void operator=(const DBIter&);
};

inline bool DBIter::ParseKey(ParsedInternalKey* ikey) {
366
  if (!ParseInternalKey(iter_.key(), ikey)) {
J
jorlow@chromium.org 已提交
367
    status_ = Status::Corruption("corrupted internal key in DBIter");
368
    valid_ = false;
369
    ROCKS_LOG_ERROR(logger_, "corrupted internal key in DBIter: %s",
370
                    iter_.key().ToString(true).c_str());
J
jorlow@chromium.org 已提交
371 372 373 374 375 376
    return false;
  } else {
    return true;
  }
}

J
jorlow@chromium.org 已提交
377 378
void DBIter::Next() {
  assert(valid_);
379
  assert(status_.ok());
J
jorlow@chromium.org 已提交
380

381
  PERF_CPU_TIMER_GUARD(iter_next_cpu_nanos, env_);
382 383
  // Release temporarily pinned blocks from last operation
  ReleaseTempPinnedData();
384 385 386
  local_stats_.skip_count_ += num_internal_keys_skipped_;
  local_stats_.skip_count_--;
  num_internal_keys_skipped_ = 0;
387
  bool ok = true;
S
Stanislau Hlebik 已提交
388
  if (direction_ == kReverse) {
389
    is_key_seqnum_zero_ = false;
390 391 392
    if (!ReverseToForward()) {
      ok = false;
    }
393
  } else if (!current_entry_is_merged_) {
394 395 396 397 398
    // 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.
399 400
    assert(iter_.Valid());
    iter_.Next();
401
    PERF_COUNTER_ADD(internal_key_skipped_count, 1);
J
jorlow@chromium.org 已提交
402
  }
J
jorlow@chromium.org 已提交
403

404
  local_stats_.next_count_++;
405
  if (ok && iter_.Valid()) {
406 407 408
    FindNextUserEntry(true /* skipping the current user key */,
                      prefix_same_as_start_);
  } else {
409
    is_key_seqnum_zero_ = false;
410 411
    valid_ = false;
  }
412 413 414 415
  if (statistics_ != nullptr && valid_) {
    local_stats_.next_found_count_++;
    local_stats_.bytes_read_ += (key().size() + value().size());
  }
J
jorlow@chromium.org 已提交
416 417
}

418 419 420 421 422 423 424
// 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
425
//       a delete marker or a sequence number higher than sequence_
426
//       saved_key_ MUST have a proper user_key before calling this function
427 428 429 430 431
//
// 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.
432
inline bool DBIter::FindNextUserEntry(bool skipping, bool prefix_check) {
433
  PERF_TIMER_GUARD(find_next_user_entry_time);
434
  return FindNextUserEntryInternal(skipping, prefix_check);
435 436 437
}

// Actual implementation of DBIter::FindNextUserEntry()
438
inline bool DBIter::FindNextUserEntryInternal(bool skipping, bool prefix_check) {
J
jorlow@chromium.org 已提交
439
  // Loop until we hit an acceptable entry to yield
440
  assert(iter_.Valid());
441
  assert(status_.ok());
J
jorlow@chromium.org 已提交
442
  assert(direction_ == kForward);
443
  current_entry_is_merged_ = false;
444 445 446 447 448 449 450 451 452 453 454

  // 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.
455
  uint64_t num_skipped = 0;
456

Y
Yi Wu 已提交
457 458
  is_blob_ = false;

J
jorlow@chromium.org 已提交
459
  do {
460 461 462
    // Will update is_key_seqnum_zero_ as soon as we parsed the current key
    // but we need to save the previous value to be used in the loop.
    bool is_prev_key_seqnum_zero = is_key_seqnum_zero_;
463
    if (!ParseKey(&ikey_)) {
464
      is_key_seqnum_zero_ = false;
465
      return false;
466
    }
467

468 469
    is_key_seqnum_zero_ = (ikey_.sequence == 0);

470
    if (iterate_upper_bound_ != nullptr && iter_.MayBeOutOfUpperBound() &&
471
        user_comparator_.Compare(ikey_.user_key, *iterate_upper_bound_) >= 0) {
472 473
      break;
    }
474

475
    if (prefix_extractor_ && prefix_check &&
476 477
        prefix_extractor_->Transform(ikey_.user_key)
                .compare(prefix_start_key_) != 0) {
478 479 480
      break;
    }

481
    if (TooManyInternalKeysSkipped()) {
482
      return false;
483 484
    }

Y
Yi Wu 已提交
485
    if (IsVisible(ikey_.sequence)) {
486 487 488 489 490 491 492
      // If the previous entry is of seqnum 0, the current entry will not
      // possibly be skipped. This condition can potentially be relaxed to
      // prev_key.seq <= ikey_.sequence. We are cautious because it will be more
      // prone to bugs causing the same user key with the same sequence number.
      if (!is_prev_key_seqnum_zero && skipping &&
          user_comparator_.Compare(ikey_.user_key, saved_key_.GetUserKey()) <=
              0) {
493 494 495
        num_skipped++;  // skip this entry
        PERF_COUNTER_ADD(internal_key_skipped_count, 1);
      } else {
496 497
        assert(!skipping || user_comparator_.Compare(
                                ikey_.user_key, saved_key_.GetUserKey()) > 0);
498
        num_skipped = 0;
499
        switch (ikey_.type) {
500 501 502 503
          case kTypeDeletion:
          case kTypeSingleDeletion:
            // Arrange to skip all upcoming entries for this key since
            // they are hidden by this deletion.
504 505 506
            // if iterartor specified start_seqnum we
            // 1) return internal key, including the type
            // 2) return ikey only if ikey.seqnum >= start_seqnum_
507
            // note that if deletion seqnum is < start_seqnum_ we
508 509 510
            // just skip it like in normal iterator.
            if (start_seqnum_ > 0 && ikey_.sequence >= start_seqnum_)  {
              saved_key_.SetInternalKey(ikey_);
511 512
              valid_ = true;
              return true;
513 514
            } else {
              saved_key_.SetUserKey(
515 516
                  ikey_.user_key, !pin_thru_lifetime_ ||
                                      !iter_.iter()->IsKeyPinned() /* copy */);
517 518
              skipping = true;
              PERF_COUNTER_ADD(internal_delete_skipped_count, 1);
519
            }
520 521
            break;
          case kTypeValue:
Y
Yi Wu 已提交
522
          case kTypeBlobIndex:
523 524 525 526 527 528 529 530
            if (start_seqnum_ > 0) {
              // we are taking incremental snapshot here
              // incremental snapshots aren't supported on DB with range deletes
              assert(!(
                (ikey_.type == kTypeBlobIndex) && (start_seqnum_ > 0)
              ));
              if (ikey_.sequence >= start_seqnum_) {
                saved_key_.SetInternalKey(ikey_);
Y
Yi Wu 已提交
531
                valid_ = true;
532
                return true;
533 534 535
              } else {
                // this key and all previous versions shouldn't be included,
                // skipping
536 537 538 539
                saved_key_.SetUserKey(
                    ikey_.user_key,
                    !pin_thru_lifetime_ ||
                        !iter_.iter()->IsKeyPinned() /* copy */);
540
                skipping = true;
Y
Yi Wu 已提交
541
              }
542
            } else {
543
              saved_key_.SetUserKey(
544 545
                  ikey_.user_key, !pin_thru_lifetime_ ||
                                      !iter_.iter()->IsKeyPinned() /* copy */);
546
              if (range_del_agg_.ShouldDelete(
547
                      ikey_, RangeDelPositioningMode::kForwardTraversal)) {
548 549 550 551 552 553 554 555 556 557 558 559
                // 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 if (ikey_.type == kTypeBlobIndex) {
                if (!allow_blob_) {
                  ROCKS_LOG_ERROR(logger_, "Encounter unexpected blob index.");
                  status_ = Status::NotSupported(
                      "Encounter unexpected blob index. Please open DB with "
                      "rocksdb::blob_db::BlobDB instead.");
                  valid_ = false;
560
                  return false;
561
                }
562 563 564 565

                is_blob_ = true;
                valid_ = true;
                return true;
566 567
              } else {
                valid_ = true;
568
                return true;
569
              }
570 571 572
            }
            break;
          case kTypeMerge:
573
            saved_key_.SetUserKey(
574
                ikey_.user_key,
575
                !pin_thru_lifetime_ || !iter_.iter()->IsKeyPinned() /* copy */);
576
            if (range_del_agg_.ShouldDelete(
577
                    ikey_, RangeDelPositioningMode::kForwardTraversal)) {
578 579 580 581 582 583 584 585 586 587
              // 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;
588
              return MergeValuesNewToOld();  // Go to a different state machine
589 590 591 592 593
            }
            break;
          default:
            assert(false);
            break;
594
        }
J
jorlow@chromium.org 已提交
595
      }
596 597 598
    } else {
      PERF_COUNTER_ADD(internal_recent_skipped_count, 1);

599 600 601 602
      // This key was inserted after our snapshot was taken.
      // If this happens too many times in a row for the same user key, we want
      // to seek to the target sequence number.
      int cmp =
603
          user_comparator_.Compare(ikey_.user_key, saved_key_.GetUserKey());
604
      if (cmp == 0 || (skipping && cmp <= 0)) {
605 606
        num_skipped++;
      } else {
607
        saved_key_.SetUserKey(
608
            ikey_.user_key,
609
            !iter_.iter()->IsKeyPinned() || !pin_thru_lifetime_ /* copy */);
610 611 612
        skipping = false;
        num_skipped = 0;
      }
J
jorlow@chromium.org 已提交
613
    }
614 615 616

    // If we have sequentially iterated via numerous equal keys, then it's
    // better to seek so that we can avoid too many key comparisons.
617
    if (num_skipped > max_skip_ && CanReseekToSkip()) {
618
      is_key_seqnum_zero_ = false;
619 620
      num_skipped = 0;
      std::string last_key;
621 622 623 624
      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).
625 626
        AppendInternalKey(&last_key, ParsedInternalKey(saved_key_.GetUserKey(),
                                                       0, kTypeDeletion));
627 628 629 630 631 632 633 634 635
        // 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,
636
                          ParsedInternalKey(saved_key_.GetUserKey(), sequence_,
637 638
                                            kValueTypeForSeek));
      }
639
      iter_.Seek(last_key);
640 641
      RecordTick(statistics_, NUMBER_OF_RESEEKS_IN_ITERATION);
    } else {
642
      iter_.Next();
643
    }
644
  } while (iter_.Valid());
645

J
jorlow@chromium.org 已提交
646
  valid_ = false;
647
  return iter_.status().ok();
J
jorlow@chromium.org 已提交
648 649
}

650 651
// Merge values of the same user key starting from the current iter_ position
// Scan from the newer entries to older entries.
652
// PRE: iter_.key() points to the first merge type entry
653 654 655
//      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)
656
bool DBIter::MergeValuesNewToOld() {
657
  if (!merge_operator_) {
658
    ROCKS_LOG_ERROR(logger_, "Options::merge_operator is null.");
659
    status_ = Status::InvalidArgument("merge_operator_ must be set.");
660
    valid_ = false;
661
    return false;
D
Deon Nicholas 已提交
662
  }
663

664 665
  // Temporarily pin the blocks that hold merge operands
  TempPinData();
666
  merge_context_.Clear();
667
  // Start the merge process by pushing the first operand
668 669
  merge_context_.PushOperand(
      iter_.value(), iter_.iter()->IsValuePinned() /* operand_pinned */);
670
  TEST_SYNC_POINT("DBIter::MergeValuesNewToOld:PushedFirstOperand");
671 672

  ParsedInternalKey ikey;
673
  Status s;
674
  for (iter_.Next(); iter_.Valid(); iter_.Next()) {
675
    TEST_SYNC_POINT("DBIter::MergeValuesNewToOld:SteppedToNextOperand");
676
    if (!ParseKey(&ikey)) {
677
      return false;
678 679
    }

680
    if (!user_comparator_.Equal(ikey.user_key, saved_key_.GetUserKey())) {
681 682
      // hit the next user key, stop right here
      break;
A
Andrew Kryczka 已提交
683
    } else if (kTypeDeletion == ikey.type || kTypeSingleDeletion == ikey.type ||
684
               range_del_agg_.ShouldDelete(
685
                   ikey, RangeDelPositioningMode::kForwardTraversal)) {
686 687
      // hit a delete with the same user key, stop right here
      // iter_ is positioned after delete
688
      iter_.Next();
689
      break;
A
Andres Noetzli 已提交
690
    } else if (kTypeValue == ikey.type) {
691 692
      // hit a put, merge the put value with operands and store the
      // final result in saved_value_. We are done!
693
      const Slice val = iter_.value();
694 695
      s = MergeHelper::TimedFullMerge(
          merge_operator_, ikey.user_key, &val, merge_context_.GetOperands(),
696
          &saved_value_, logger_, statistics_, env_, &pinned_value_, true);
697
      if (!s.ok()) {
Y
Yi Wu 已提交
698
        valid_ = false;
699
        status_ = s;
700
        return false;
701
      }
702
      // iter_ is positioned after put
703 704
      iter_.Next();
      if (!iter_.status().ok()) {
705 706 707 708
        valid_ = false;
        return false;
      }
      return true;
A
Andres Noetzli 已提交
709
    } else if (kTypeMerge == ikey.type) {
710 711
      // hit a merge, add the value as an operand and run associative merge.
      // when complete, add result to operands and continue.
712 713
      merge_context_.PushOperand(
          iter_.value(), iter_.iter()->IsValuePinned() /* operand_pinned */);
714
      PERF_COUNTER_ADD(internal_merge_count, 1);
Y
Yi Wu 已提交
715 716 717 718 719 720 721 722 723 724 725
    } else if (kTypeBlobIndex == ikey.type) {
      if (!allow_blob_) {
        ROCKS_LOG_ERROR(logger_, "Encounter unexpected blob index.");
        status_ = Status::NotSupported(
            "Encounter unexpected blob index. Please open DB with "
            "rocksdb::blob_db::BlobDB instead.");
      } else {
        status_ =
            Status::NotSupported("Blob DB does not support merge operator.");
      }
      valid_ = false;
726
      return false;
A
Andres Noetzli 已提交
727 728
    } else {
      assert(false);
729 730 731
    }
  }

732
  if (!iter_.status().ok()) {
733 734 735 736
    valid_ = false;
    return false;
  }

737 738 739 740
  // 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.
741 742 743
  s = MergeHelper::TimedFullMerge(merge_operator_, saved_key_.GetUserKey(),
                                  nullptr, merge_context_.GetOperands(),
                                  &saved_value_, logger_, statistics_, env_,
744
                                  &pinned_value_, true);
745
  if (!s.ok()) {
Y
Yi Wu 已提交
746
    valid_ = false;
747
    status_ = s;
748
    return false;
749
  }
750 751 752

  assert(status_.ok());
  return true;
753 754
}

J
jorlow@chromium.org 已提交
755 756
void DBIter::Prev() {
  assert(valid_);
757
  assert(status_.ok());
758 759

  PERF_CPU_TIMER_GUARD(iter_prev_cpu_nanos, env_);
760
  ReleaseTempPinnedData();
761
  ResetInternalKeysSkippedCounter();
762
  bool ok = true;
S
Stanislau Hlebik 已提交
763
  if (direction_ == kForward) {
764 765 766 767 768 769
    if (!ReverseToBackward()) {
      ok = false;
    }
  }
  if (ok) {
    PrevInternal();
S
Stanislau Hlebik 已提交
770
  }
M
Manuel Ung 已提交
771
  if (statistics_ != nullptr) {
772
    local_stats_.prev_count_++;
M
Manuel Ung 已提交
773
    if (valid_) {
774 775
      local_stats_.prev_found_count_++;
      local_stats_.bytes_read_ += (key().size() + value().size());
M
Manuel Ung 已提交
776 777
    }
  }
S
Stanislau Hlebik 已提交
778
}
J
jorlow@chromium.org 已提交
779

780
bool DBIter::ReverseToForward() {
781
  assert(iter_.status().ok());
782 783 784 785

  // When moving backwards, iter_ is positioned on _previous_ key, which may
  // not exist or may have different prefix than the current key().
  // If that's the case, seek iter_ to current key.
786
  if ((prefix_extractor_ != nullptr && !total_order_seek_) || !iter_.Valid()) {
787 788
    IterKey last_key;
    last_key.SetInternalKey(ParsedInternalKey(
789
        saved_key_.GetUserKey(), kMaxSequenceNumber, kValueTypeForSeek));
790
    iter_.Seek(last_key.GetInternalKey());
791
  }
792

793
  direction_ = kForward;
794
  // Skip keys less than the current key() (a.k.a. saved_key_).
795
  while (iter_.Valid()) {
796 797 798 799
    ParsedInternalKey ikey;
    if (!ParseKey(&ikey)) {
      return false;
    }
800
    if (user_comparator_.Compare(ikey.user_key, saved_key_.GetUserKey()) >= 0) {
801 802
      return true;
    }
803
    iter_.Next();
804 805
  }

806
  if (!iter_.status().ok()) {
807 808
    valid_ = false;
    return false;
809
  }
810 811

  return true;
812 813
}

814 815
// Move iter_ to the key before saved_key_.
bool DBIter::ReverseToBackward() {
816
  assert(iter_.status().ok());
817 818 819 820 821 822

  // When current_entry_is_merged_ is true, iter_ may be positioned on the next
  // key, which may not exist or may have prefix different from current.
  // If that's the case, seek to saved_key_.
  if (current_entry_is_merged_ &&
      ((prefix_extractor_ != nullptr && !total_order_seek_) ||
823
       !iter_.Valid())) {
824
    IterKey last_key;
825 826 827 828 829 830
    // Using kMaxSequenceNumber and kValueTypeForSeek
    // (not kValueTypeForSeekForPrev) to seek to a key strictly smaller
    // than saved_key_.
    last_key.SetInternalKey(ParsedInternalKey(
        saved_key_.GetUserKey(), kMaxSequenceNumber, kValueTypeForSeek));
    if (prefix_extractor_ != nullptr && !total_order_seek_) {
831
      iter_.SeekForPrev(last_key.GetInternalKey());
832 833 834 835 836
    } else {
      // Some iterators may not support SeekForPrev(), so we avoid using it
      // when prefix seek mode is disabled. This is somewhat expensive
      // (an extra Prev(), as well as an extra change of direction of iter_),
      // so we may need to reconsider it later.
837 838 839
      iter_.Seek(last_key.GetInternalKey());
      if (!iter_.Valid() && iter_.status().ok()) {
        iter_.SeekToLast();
840
      }
841 842 843 844
    }
  }

  direction_ = kReverse;
845
  return FindUserKeyBeforeSavedKey();
846 847
}

S
Stanislau Hlebik 已提交
848
void DBIter::PrevInternal() {
849
  while (iter_.Valid()) {
850
    saved_key_.SetUserKey(
851 852
        ExtractUserKey(iter_.key()),
        !iter_.iter()->IsKeyPinned() || !pin_thru_lifetime_ /* copy */);
853

854 855 856 857 858 859 860 861
    if (prefix_extractor_ && prefix_same_as_start_ &&
        prefix_extractor_->Transform(saved_key_.GetUserKey())
                .compare(prefix_start_key_) != 0) {
      // Current key does not have the same prefix as start
      valid_ = false;
      return;
    }

862
    if (iterate_lower_bound_ != nullptr && iter_.MayBeOutOfLowerBound() &&
863 864
        user_comparator_.Compare(saved_key_.GetUserKey(),
                                 *iterate_lower_bound_) < 0) {
865 866 867 868 869
      // We've iterated earlier than the user-specified lower bound.
      valid_ = false;
      return;
    }

870
    if (!FindValueForCurrentKey()) {  // assigns valid_
S
Stanislau Hlebik 已提交
871
      return;
J
jorlow@chromium.org 已提交
872
    }
873

874 875 876
    // Whether or not we found a value for current key, we need iter_ to end up
    // on a smaller key.
    if (!FindUserKeyBeforeSavedKey()) {
877 878 879
      return;
    }

880 881 882
    if (valid_) {
      // Found the value.
      return;
S
Stanislau Hlebik 已提交
883
    }
884 885 886

    if (TooManyInternalKeysSkipped(false)) {
      return;
S
Stanislau Hlebik 已提交
887 888
    }
  }
889

S
Stanislau Hlebik 已提交
890 891
  // We haven't found any key - iterator is not valid
  valid_ = false;
J
jorlow@chromium.org 已提交
892 893
}

894 895 896 897 898 899 900 901 902 903 904
// Used for backwards iteration.
// Looks at the entries with user key saved_key_ and finds the most up-to-date
// value for it, or executes a merge, or determines that the value was deleted.
// Sets valid_ to true if the value is found and is ready to be presented to
// the user through value().
// Sets valid_ to false if the value was deleted, and we should try another key.
// Returns false if an error occurred, and !status().ok() and !valid_.
//
// PRE: iter_ is positioned on the last entry with user key equal to saved_key_.
// POST: iter_ is positioned on one of the entries equal to saved_key_, or on
//       the entry just before them, or on the entry just after them.
S
Stanislau Hlebik 已提交
905
bool DBIter::FindValueForCurrentKey() {
906
  assert(iter_.Valid());
907
  merge_context_.Clear();
908
  current_entry_is_merged_ = false;
A
Andres Noetzli 已提交
909 910
  // last entry before merge (could be kTypeDeletion, kTypeSingleDeletion or
  // kTypeValue)
S
Stanislau Hlebik 已提交
911 912
  ValueType last_not_merge_type = kTypeDeletion;
  ValueType last_key_entry_type = kTypeDeletion;
J
jorlow@chromium.org 已提交
913

914 915 916
  // Temporarily pin blocks that hold (merge operands / the value)
  ReleaseTempPinnedData();
  TempPinData();
S
Stanislau Hlebik 已提交
917
  size_t num_skipped = 0;
918
  while (iter_.Valid()) {
919 920 921 922 923 924
    ParsedInternalKey ikey;
    if (!ParseKey(&ikey)) {
      return false;
    }

    if (!IsVisible(ikey.sequence) ||
925
        !user_comparator_.Equal(ikey.user_key, saved_key_.GetUserKey())) {
926 927
      break;
    }
928 929 930 931
    if (TooManyInternalKeysSkipped()) {
      return false;
    }

932 933 934
    // This user key has lots of entries.
    // We're going from old to new, and it's taking too long. Let's do a Seek()
    // and go from new to old. This helps when a key was overwritten many times.
935
    if (num_skipped >= max_skip_ && CanReseekToSkip()) {
S
Stanislau Hlebik 已提交
936 937 938 939 940 941
      return FindValueForCurrentKeyUsingSeek();
    }

    last_key_entry_type = ikey.type;
    switch (last_key_entry_type) {
      case kTypeValue:
Y
Yi Wu 已提交
942
      case kTypeBlobIndex:
943
        if (range_del_agg_.ShouldDelete(
944
                ikey, RangeDelPositioningMode::kBackwardTraversal)) {
A
Andrew Kryczka 已提交
945 946 947
          last_key_entry_type = kTypeRangeDeletion;
          PERF_COUNTER_ADD(internal_delete_skipped_count, 1);
        } else {
948 949
          assert(iter_.iter()->IsValuePinned());
          pinned_value_ = iter_.value();
A
Andrew Kryczka 已提交
950
        }
951
        merge_context_.Clear();
A
Andrew Kryczka 已提交
952
        last_not_merge_type = last_key_entry_type;
S
Stanislau Hlebik 已提交
953 954
        break;
      case kTypeDeletion:
A
Andres Noetzli 已提交
955
      case kTypeSingleDeletion:
956
        merge_context_.Clear();
A
Andres Noetzli 已提交
957
        last_not_merge_type = last_key_entry_type;
958
        PERF_COUNTER_ADD(internal_delete_skipped_count, 1);
S
Stanislau Hlebik 已提交
959 960
        break;
      case kTypeMerge:
961
        if (range_del_agg_.ShouldDelete(
962
                ikey, RangeDelPositioningMode::kBackwardTraversal)) {
A
Andrew Kryczka 已提交
963 964 965 966 967 968 969
          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(
970 971
              iter_.value(),
              iter_.iter()->IsValuePinned() /* operand_pinned */);
972
          PERF_COUNTER_ADD(internal_merge_count, 1);
A
Andrew Kryczka 已提交
973
        }
S
Stanislau Hlebik 已提交
974 975 976 977 978
        break;
      default:
        assert(false);
    }

979
    PERF_COUNTER_ADD(internal_key_skipped_count, 1);
980
    iter_.Prev();
S
Stanislau Hlebik 已提交
981
    ++num_skipped;
982 983
  }

984
  if (!iter_.status().ok()) {
985 986
    valid_ = false;
    return false;
S
Stanislau Hlebik 已提交
987 988
  }

989
  Status s;
Y
Yi Wu 已提交
990
  is_blob_ = false;
S
Stanislau Hlebik 已提交
991 992
  switch (last_key_entry_type) {
    case kTypeDeletion:
A
Andres Noetzli 已提交
993
    case kTypeSingleDeletion:
A
Andrew Kryczka 已提交
994
    case kTypeRangeDeletion:
S
Stanislau Hlebik 已提交
995
      valid_ = false;
996
      return true;
S
Stanislau Hlebik 已提交
997
    case kTypeMerge:
998
      current_entry_is_merged_ = true;
A
Aaron Gao 已提交
999
      if (last_not_merge_type == kTypeDeletion ||
A
Andrew Kryczka 已提交
1000 1001
          last_not_merge_type == kTypeSingleDeletion ||
          last_not_merge_type == kTypeRangeDeletion) {
1002 1003 1004
        s = MergeHelper::TimedFullMerge(
            merge_operator_, saved_key_.GetUserKey(), nullptr,
            merge_context_.GetOperands(), &saved_value_, logger_, statistics_,
1005
            env_, &pinned_value_, true);
Y
Yi Wu 已提交
1006 1007 1008 1009 1010 1011 1012 1013 1014 1015 1016
      } else if (last_not_merge_type == kTypeBlobIndex) {
        if (!allow_blob_) {
          ROCKS_LOG_ERROR(logger_, "Encounter unexpected blob index.");
          status_ = Status::NotSupported(
              "Encounter unexpected blob index. Please open DB with "
              "rocksdb::blob_db::BlobDB instead.");
        } else {
          status_ =
              Status::NotSupported("Blob DB does not support merge operator.");
        }
        valid_ = false;
1017
        return false;
1018
      } else {
S
Stanislau Hlebik 已提交
1019
        assert(last_not_merge_type == kTypeValue);
1020
        s = MergeHelper::TimedFullMerge(
1021
            merge_operator_, saved_key_.GetUserKey(), &pinned_value_,
1022
            merge_context_.GetOperands(), &saved_value_, logger_, statistics_,
1023
            env_, &pinned_value_, true);
1024
      }
S
Stanislau Hlebik 已提交
1025 1026
      break;
    case kTypeValue:
1027
      // do nothing - we've already has value in pinned_value_
S
Stanislau Hlebik 已提交
1028
      break;
Y
Yi Wu 已提交
1029 1030 1031 1032 1033 1034 1035
    case kTypeBlobIndex:
      if (!allow_blob_) {
        ROCKS_LOG_ERROR(logger_, "Encounter unexpected blob index.");
        status_ = Status::NotSupported(
            "Encounter unexpected blob index. Please open DB with "
            "rocksdb::blob_db::BlobDB instead.");
        valid_ = false;
1036
        return false;
Y
Yi Wu 已提交
1037 1038 1039
      }
      is_blob_ = true;
      break;
S
Stanislau Hlebik 已提交
1040 1041 1042
    default:
      assert(false);
      break;
J
jorlow@chromium.org 已提交
1043
  }
1044
  if (!s.ok()) {
Y
Yi Wu 已提交
1045
    valid_ = false;
1046
    status_ = s;
1047
    return false;
1048
  }
1049
  valid_ = true;
S
Stanislau Hlebik 已提交
1050 1051
  return true;
}
J
jorlow@chromium.org 已提交
1052

S
Stanislau Hlebik 已提交
1053 1054
// This function is used in FindValueForCurrentKey.
// We use Seek() function instead of Prev() to find necessary value
1055 1056
// TODO: This is very similar to FindNextUserEntry() and MergeValuesNewToOld().
//       Would be nice to reuse some code.
S
Stanislau Hlebik 已提交
1057
bool DBIter::FindValueForCurrentKeyUsingSeek() {
1058 1059 1060
  // FindValueForCurrentKey will enable pinning before calling
  // FindValueForCurrentKeyUsingSeek()
  assert(pinned_iters_mgr_.PinningEnabled());
S
Stanislau Hlebik 已提交
1061
  std::string last_key;
1062 1063
  AppendInternalKey(&last_key, ParsedInternalKey(saved_key_.GetUserKey(),
                                                 sequence_, kValueTypeForSeek));
1064
  iter_.Seek(last_key);
S
Stanislau Hlebik 已提交
1065 1066
  RecordTick(statistics_, NUMBER_OF_RESEEKS_IN_ITERATION);

1067 1068
  // In case read_callback presents, the value we seek to may not be visible.
  // Find the next value that's visible.
S
Stanislau Hlebik 已提交
1069
  ParsedInternalKey ikey;
1070
  while (true) {
1071
    if (!iter_.Valid()) {
1072
      valid_ = false;
1073
      return iter_.status().ok();
1074 1075 1076 1077 1078
    }

    if (!ParseKey(&ikey)) {
      return false;
    }
1079
    if (!user_comparator_.Equal(ikey.user_key, saved_key_.GetUserKey())) {
1080 1081 1082 1083 1084 1085 1086 1087 1088 1089
      // No visible values for this key, even though FindValueForCurrentKey()
      // has seen some. This is possible if we're using a tailing iterator, and
      // the entries were discarded in a compaction.
      valid_ = false;
      return true;
    }

    if (IsVisible(ikey.sequence)) {
      break;
    }
1090

1091
    iter_.Next();
1092
  }
S
Stanislau Hlebik 已提交
1093

A
Andrew Kryczka 已提交
1094
  if (ikey.type == kTypeDeletion || ikey.type == kTypeSingleDeletion ||
1095
      range_del_agg_.ShouldDelete(
1096
          ikey, RangeDelPositioningMode::kBackwardTraversal)) {
J
jorlow@chromium.org 已提交
1097
    valid_ = false;
1098
    return true;
S
Stanislau Hlebik 已提交
1099
  }
Y
Yi Wu 已提交
1100 1101 1102 1103 1104 1105
  if (ikey.type == kTypeBlobIndex && !allow_blob_) {
    ROCKS_LOG_ERROR(logger_, "Encounter unexpected blob index.");
    status_ = Status::NotSupported(
        "Encounter unexpected blob index. Please open DB with "
        "rocksdb::blob_db::BlobDB instead.");
    valid_ = false;
1106
    return false;
Y
Yi Wu 已提交
1107 1108
  }
  if (ikey.type == kTypeValue || ikey.type == kTypeBlobIndex) {
1109 1110
    assert(iter_.iter()->IsValuePinned());
    pinned_value_ = iter_.value();
A
Andrew Kryczka 已提交
1111 1112 1113
    valid_ = true;
    return true;
  }
S
Stanislau Hlebik 已提交
1114 1115 1116

  // kTypeMerge. We need to collect all kTypeMerge values and save them
  // in operands
1117
  assert(ikey.type == kTypeMerge);
1118
  current_entry_is_merged_ = true;
1119
  merge_context_.Clear();
1120 1121
  merge_context_.PushOperand(
      iter_.value(), iter_.iter()->IsValuePinned() /* operand_pinned */);
1122
  while (true) {
1123
    iter_.Next();
S
Stanislau Hlebik 已提交
1124

1125 1126
    if (!iter_.Valid()) {
      if (!iter_.status().ok()) {
1127 1128 1129 1130
        valid_ = false;
        return false;
      }
      break;
S
Stanislau Hlebik 已提交
1131
    }
1132 1133 1134
    if (!ParseKey(&ikey)) {
      return false;
    }
1135
    if (!user_comparator_.Equal(ikey.user_key, saved_key_.GetUserKey())) {
1136 1137 1138 1139 1140
      break;
    }

    if (ikey.type == kTypeDeletion || ikey.type == kTypeSingleDeletion ||
        range_del_agg_.ShouldDelete(
1141
            ikey, RangeDelPositioningMode::kForwardTraversal)) {
1142 1143
      break;
    } else if (ikey.type == kTypeValue) {
1144
      const Slice val = iter_.value();
1145 1146 1147 1148 1149 1150 1151 1152 1153
      Status s = MergeHelper::TimedFullMerge(
          merge_operator_, saved_key_.GetUserKey(), &val,
          merge_context_.GetOperands(), &saved_value_, logger_, statistics_,
          env_, &pinned_value_, true);
      if (!s.ok()) {
        valid_ = false;
        status_ = s;
        return false;
      }
Y
Yi Wu 已提交
1154
      valid_ = true;
1155 1156
      return true;
    } else if (ikey.type == kTypeMerge) {
1157 1158
      merge_context_.PushOperand(
          iter_.value(), iter_.iter()->IsValuePinned() /* operand_pinned */);
1159 1160 1161 1162 1163 1164 1165 1166 1167 1168 1169
      PERF_COUNTER_ADD(internal_merge_count, 1);
    } else if (ikey.type == kTypeBlobIndex) {
      if (!allow_blob_) {
        ROCKS_LOG_ERROR(logger_, "Encounter unexpected blob index.");
        status_ = Status::NotSupported(
            "Encounter unexpected blob index. Please open DB with "
            "rocksdb::blob_db::BlobDB instead.");
      } else {
        status_ =
            Status::NotSupported("Blob DB does not support merge operator.");
      }
Y
Yi Wu 已提交
1170
      valid_ = false;
1171 1172 1173
      return false;
    } else {
      assert(false);
1174
    }
S
Stanislau Hlebik 已提交
1175 1176
  }

1177 1178 1179 1180 1181
  Status s = MergeHelper::TimedFullMerge(
      merge_operator_, saved_key_.GetUserKey(), nullptr,
      merge_context_.GetOperands(), &saved_value_, logger_, statistics_, env_,
      &pinned_value_, true);
  if (!s.ok()) {
Y
Yi Wu 已提交
1182
    valid_ = false;
1183
    status_ = s;
1184
    return false;
1185
  }
S
Stanislau Hlebik 已提交
1186

1187 1188 1189
  // Make sure we leave iter_ in a good state. If it's valid and we don't care
  // about prefixes, that's already good enough. Otherwise it needs to be
  // seeked to the current key.
1190
  if ((prefix_extractor_ != nullptr && !total_order_seek_) || !iter_.Valid()) {
1191
    if (prefix_extractor_ != nullptr && !total_order_seek_) {
1192
      iter_.SeekForPrev(last_key);
1193
    } else {
1194 1195 1196
      iter_.Seek(last_key);
      if (!iter_.Valid() && iter_.status().ok()) {
        iter_.SeekToLast();
1197 1198 1199
      }
    }
    RecordTick(statistics_, NUMBER_OF_RESEEKS_IN_ITERATION);
S
Stanislau Hlebik 已提交
1200
  }
1201 1202 1203

  valid_ = true;
  return true;
S
Stanislau Hlebik 已提交
1204 1205
}

1206 1207 1208 1209
// Move backwards until the key smaller than saved_key_.
// Changes valid_ only if return value is false.
bool DBIter::FindUserKeyBeforeSavedKey() {
  assert(status_.ok());
S
Stanislau Hlebik 已提交
1210
  size_t num_skipped = 0;
1211
  while (iter_.Valid()) {
1212 1213 1214
    ParsedInternalKey ikey;
    if (!ParseKey(&ikey)) {
      return false;
1215 1216
    }

1217
    if (user_comparator_.Compare(ikey.user_key, saved_key_.GetUserKey()) < 0) {
1218 1219 1220 1221 1222
      return true;
    }

    if (TooManyInternalKeysSkipped()) {
      return false;
S
Stanislau Hlebik 已提交
1223
    }
1224

S
Siying Dong 已提交
1225
    assert(ikey.sequence != kMaxSequenceNumber);
Y
Yi Wu 已提交
1226
    if (!IsVisible(ikey.sequence)) {
1227 1228 1229 1230
      PERF_COUNTER_ADD(internal_recent_skipped_count, 1);
    } else {
      PERF_COUNTER_ADD(internal_key_skipped_count, 1);
    }
1231

1232
    if (num_skipped >= max_skip_ && CanReseekToSkip()) {
1233 1234 1235 1236 1237 1238
      num_skipped = 0;
      IterKey last_key;
      last_key.SetInternalKey(ParsedInternalKey(
          saved_key_.GetUserKey(), kMaxSequenceNumber, kValueTypeForSeek));
      // It would be more efficient to use SeekForPrev() here, but some
      // iterators may not support it.
1239
      iter_.Seek(last_key.GetInternalKey());
1240
      RecordTick(statistics_, NUMBER_OF_RESEEKS_IN_ITERATION);
1241
      if (!iter_.Valid()) {
1242 1243 1244 1245 1246 1247
        break;
      }
    } else {
      ++num_skipped;
    }

1248
    iter_.Prev();
S
Stanislau Hlebik 已提交
1249
  }
1250

1251
  if (!iter_.status().ok()) {
1252 1253 1254 1255 1256
    valid_ = false;
    return false;
  }

  return true;
S
Stanislau Hlebik 已提交
1257 1258
}

1259 1260 1261 1262 1263 1264 1265 1266 1267 1268 1269 1270
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;
}

Y
Yi Wu 已提交
1271
bool DBIter::IsVisible(SequenceNumber sequence) {
1272
  if (read_callback_ == nullptr) {
1273 1274 1275
    return sequence <= sequence_;
  } else {
    return read_callback_->IsVisible(sequence);
1276
  }
1277
}
1278

1279 1280
bool DBIter::CanReseekToSkip() {
  return read_callback_ == nullptr || read_callback_->CanReseekToSkip();
Y
Yi Wu 已提交
1281 1282
}

J
jorlow@chromium.org 已提交
1283
void DBIter::Seek(const Slice& target) {
1284
  PERF_CPU_TIMER_GUARD(iter_seek_cpu_nanos, env_);
L
Lei Jin 已提交
1285
  StopWatch sw(env_, statistics_, DB_SEEK);
1286
  status_ = Status::OK();
1287
  ReleaseTempPinnedData();
1288
  ResetInternalKeysSkippedCounter();
1289
  is_key_seqnum_zero_ = false;
1290

1291
  SequenceNumber seq = sequence_;
1292
  saved_key_.Clear();
1293
  saved_key_.SetInternalKey(target, seq);
1294

1295 1296 1297 1298 1299 1300
#ifndef ROCKSDB_LITE
  if (db_impl_ != nullptr && cfd_ != nullptr) {
    db_impl_->TraceIteratorSeek(cfd_->GetID(), target);
  }
#endif  // ROCKSDB_LITE

Z
zhangjinpeng1987 已提交
1301
  if (iterate_lower_bound_ != nullptr &&
1302 1303
      user_comparator_.Compare(saved_key_.GetUserKey(), *iterate_lower_bound_) <
          0) {
Z
zhangjinpeng1987 已提交
1304
    saved_key_.Clear();
1305
    saved_key_.SetInternalKey(*iterate_lower_bound_, seq);
Z
zhangjinpeng1987 已提交
1306 1307
  }

1308 1309
  {
    PERF_TIMER_GUARD(seek_internal_seek_time);
1310
    iter_.Seek(saved_key_.GetInternalKey());
1311
    range_del_agg_.InvalidateRangeDelMapPositions();
1312
  }
M
Manuel Ung 已提交
1313
  RecordTick(statistics_, NUMBER_DB_SEEK);
1314
  if (iter_.Valid()) {
1315 1316 1317
    if (prefix_extractor_ && prefix_same_as_start_) {
      prefix_start_key_ = prefix_extractor_->Transform(target);
    }
1318 1319
    direction_ = kForward;
    ClearSavedValue();
1320 1321 1322 1323
    FindNextUserEntry(false /* not skipping */, prefix_same_as_start_);
    if (!valid_) {
      prefix_start_key_.clear();
    }
M
Manuel Ung 已提交
1324 1325
    if (statistics_ != nullptr) {
      if (valid_) {
1326
        // Decrement since we don't want to count this key as skipped
M
Manuel Ung 已提交
1327 1328
        RecordTick(statistics_, NUMBER_DB_SEEK_FOUND);
        RecordTick(statistics_, ITER_BYTES_READ, key().size() + value().size());
1329
        PERF_COUNTER_ADD(iter_read_bytes, key().size() + value().size());
M
Manuel Ung 已提交
1330 1331
      }
    }
J
jorlow@chromium.org 已提交
1332 1333 1334
  } else {
    valid_ = false;
  }
1335

1336
  if (valid_ && prefix_extractor_ && prefix_same_as_start_) {
1337 1338
    prefix_start_buf_.SetUserKey(prefix_start_key_);
    prefix_start_key_ = prefix_start_buf_.GetUserKey();
1339
  }
J
jorlow@chromium.org 已提交
1340
}
J
jorlow@chromium.org 已提交
1341

A
Aaron Gao 已提交
1342
void DBIter::SeekForPrev(const Slice& target) {
1343
  PERF_CPU_TIMER_GUARD(iter_seek_cpu_nanos, env_);
A
Aaron Gao 已提交
1344
  StopWatch sw(env_, statistics_, DB_SEEK);
1345
  status_ = Status::OK();
A
Aaron Gao 已提交
1346
  ReleaseTempPinnedData();
1347
  ResetInternalKeysSkippedCounter();
1348
  is_key_seqnum_zero_ = false;
A
Aaron Gao 已提交
1349 1350 1351 1352 1353
  saved_key_.Clear();
  // now saved_key is used to store internal key.
  saved_key_.SetInternalKey(target, 0 /* sequence_number */,
                            kValueTypeForSeekForPrev);

Z
zhangjinpeng1987 已提交
1354
  if (iterate_upper_bound_ != nullptr &&
1355 1356
      user_comparator_.Compare(saved_key_.GetUserKey(),
                               *iterate_upper_bound_) >= 0) {
Z
zhangjinpeng1987 已提交
1357 1358 1359 1360
    saved_key_.Clear();
    saved_key_.SetInternalKey(*iterate_upper_bound_, kMaxSequenceNumber);
  }

A
Aaron Gao 已提交
1361 1362
  {
    PERF_TIMER_GUARD(seek_internal_seek_time);
1363
    iter_.SeekForPrev(saved_key_.GetInternalKey());
1364
    range_del_agg_.InvalidateRangeDelMapPositions();
A
Aaron Gao 已提交
1365 1366
  }

1367 1368 1369 1370 1371 1372
#ifndef ROCKSDB_LITE
  if (db_impl_ != nullptr && cfd_ != nullptr) {
    db_impl_->TraceIteratorSeekForPrev(cfd_->GetID(), target);
  }
#endif  // ROCKSDB_LITE

A
Aaron Gao 已提交
1373
  RecordTick(statistics_, NUMBER_DB_SEEK);
1374
  if (iter_.Valid()) {
A
Aaron Gao 已提交
1375 1376 1377 1378 1379 1380 1381 1382 1383 1384 1385 1386 1387
    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());
1388
        PERF_COUNTER_ADD(iter_read_bytes, key().size() + value().size());
A
Aaron Gao 已提交
1389 1390 1391 1392 1393 1394
      }
    }
  } else {
    valid_ = false;
  }
  if (valid_ && prefix_extractor_ && prefix_same_as_start_) {
1395 1396
    prefix_start_buf_.SetUserKey(prefix_start_key_);
    prefix_start_key_ = prefix_start_buf_.GetUserKey();
A
Aaron Gao 已提交
1397 1398 1399
  }
}

J
jorlow@chromium.org 已提交
1400
void DBIter::SeekToFirst() {
1401 1402 1403 1404
  if (iterate_lower_bound_ != nullptr) {
    Seek(*iterate_lower_bound_);
    return;
  }
1405
  PERF_CPU_TIMER_GUARD(iter_seek_cpu_nanos, env_);
1406 1407 1408 1409 1410 1411
  // Don't use iter_::Seek() if we set a prefix extractor
  // because prefix seek will be used.
  if (prefix_extractor_ != nullptr && !total_order_seek_) {
    max_skip_ = std::numeric_limits<uint64_t>::max();
  }
  status_ = Status::OK();
J
jorlow@chromium.org 已提交
1412
  direction_ = kForward;
1413
  ReleaseTempPinnedData();
1414
  ResetInternalKeysSkippedCounter();
J
jorlow@chromium.org 已提交
1415
  ClearSavedValue();
1416
  is_key_seqnum_zero_ = false;
1417 1418 1419

  {
    PERF_TIMER_GUARD(seek_internal_seek_time);
1420
    iter_.SeekToFirst();
1421
    range_del_agg_.InvalidateRangeDelMapPositions();
1422 1423
  }

M
Manuel Ung 已提交
1424
  RecordTick(statistics_, NUMBER_DB_SEEK);
1425
  if (iter_.Valid()) {
1426
    saved_key_.SetUserKey(
1427 1428
        ExtractUserKey(iter_.key()),
        !iter_.iter()->IsKeyPinned() || !pin_thru_lifetime_ /* copy */);
1429
    FindNextUserEntry(false /* not skipping */, false /* no prefix check */);
M
Manuel Ung 已提交
1430 1431 1432 1433
    if (statistics_ != nullptr) {
      if (valid_) {
        RecordTick(statistics_, NUMBER_DB_SEEK_FOUND);
        RecordTick(statistics_, ITER_BYTES_READ, key().size() + value().size());
1434
        PERF_COUNTER_ADD(iter_read_bytes, key().size() + value().size());
M
Manuel Ung 已提交
1435 1436
      }
    }
J
jorlow@chromium.org 已提交
1437 1438
  } else {
    valid_ = false;
J
jorlow@chromium.org 已提交
1439
  }
1440
  if (valid_ && prefix_extractor_ && prefix_same_as_start_) {
1441 1442 1443
    prefix_start_buf_.SetUserKey(
        prefix_extractor_->Transform(saved_key_.GetUserKey()));
    prefix_start_key_ = prefix_start_buf_.GetUserKey();
1444
  }
J
jorlow@chromium.org 已提交
1445 1446
}

J
jorlow@chromium.org 已提交
1447
void DBIter::SeekToLast() {
1448 1449 1450
  if (iterate_upper_bound_ != nullptr) {
    // Seek to last key strictly less than ReadOptions.iterate_upper_bound.
    SeekForPrev(*iterate_upper_bound_);
1451
    if (Valid() && user_comparator_.Equal(*iterate_upper_bound_, key())) {
1452 1453 1454 1455 1456 1457
      ReleaseTempPinnedData();
      PrevInternal();
    }
    return;
  }

1458
  PERF_CPU_TIMER_GUARD(iter_seek_cpu_nanos, env_);
S
Stanislau Hlebik 已提交
1459
  // Don't use iter_::Seek() if we set a prefix extractor
1460
  // because prefix seek will be used.
1461
  if (prefix_extractor_ != nullptr && !total_order_seek_) {
S
Stanislau Hlebik 已提交
1462 1463
    max_skip_ = std::numeric_limits<uint64_t>::max();
  }
1464
  status_ = Status::OK();
J
jorlow@chromium.org 已提交
1465
  direction_ = kReverse;
1466
  ReleaseTempPinnedData();
1467
  ResetInternalKeysSkippedCounter();
J
jorlow@chromium.org 已提交
1468
  ClearSavedValue();
1469
  is_key_seqnum_zero_ = false;
1470 1471 1472

  {
    PERF_TIMER_GUARD(seek_internal_seek_time);
1473
    iter_.SeekToLast();
1474
    range_del_agg_.InvalidateRangeDelMapPositions();
1475
  }
1476
  PrevInternal();
M
Manuel Ung 已提交
1477 1478 1479 1480 1481
  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());
1482
      PERF_COUNTER_ADD(iter_read_bytes, key().size() + value().size());
M
Manuel Ung 已提交
1483 1484
    }
  }
1485
  if (valid_ && prefix_extractor_ && prefix_same_as_start_) {
1486 1487 1488
    prefix_start_buf_.SetUserKey(
        prefix_extractor_->Transform(saved_key_.GetUserKey()));
    prefix_start_key_ = prefix_start_buf_.GetUserKey();
1489
  }
J
jorlow@chromium.org 已提交
1490 1491
}

1492 1493
Iterator* NewDBIterator(Env* env, const ReadOptions& read_options,
                        const ImmutableCFOptions& cf_options,
1494
                        const MutableCFOptions& mutable_cf_options,
1495 1496 1497
                        const Comparator* user_key_comparator,
                        InternalIterator* internal_iter,
                        const SequenceNumber& sequence,
Y
Yi Wu 已提交
1498
                        uint64_t max_sequential_skip_in_iterations,
1499 1500 1501 1502 1503 1504
                        ReadCallback* read_callback, DBImpl* db_impl,
                        ColumnFamilyData* cfd, bool allow_blob) {
  DBIter* db_iter = new DBIter(
      env, read_options, cf_options, mutable_cf_options, user_key_comparator,
      internal_iter, sequence, false, max_sequential_skip_in_iterations,
      read_callback, db_impl, cfd, allow_blob);
1505
  return db_iter;
1506 1507
}

I
Igor Canadi 已提交
1508
ArenaWrappedDBIter::~ArenaWrappedDBIter() { db_iter_->~DBIter(); }
1509

1510
ReadRangeDelAggregator* ArenaWrappedDBIter::GetRangeDelAggregator() {
A
Andrew Kryczka 已提交
1511 1512 1513
  return db_iter_->GetRangeDelAggregator();
}

S
sdong 已提交
1514
void ArenaWrappedDBIter::SetIterUnderDBIter(InternalIterator* iter) {
1515 1516 1517 1518 1519 1520 1521 1522 1523
  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);
}
A
Aaron Gao 已提交
1524 1525 1526
inline void ArenaWrappedDBIter::SeekForPrev(const Slice& target) {
  db_iter_->SeekForPrev(target);
}
1527 1528 1529 1530 1531
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(); }
Y
Yi Wu 已提交
1532
bool ArenaWrappedDBIter::IsBlob() const { return db_iter_->IsBlob(); }
1533 1534
inline Status ArenaWrappedDBIter::GetProperty(std::string prop_name,
                                              std::string* prop) {
S
Siying Dong 已提交
1535 1536 1537 1538 1539 1540 1541
  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();
  }
1542 1543
  return db_iter_->GetProperty(prop_name, prop);
}
S
Siying Dong 已提交
1544 1545 1546

void ArenaWrappedDBIter::Init(Env* env, const ReadOptions& read_options,
                              const ImmutableCFOptions& cf_options,
1547
                              const MutableCFOptions& mutable_cf_options,
S
Siying Dong 已提交
1548 1549
                              const SequenceNumber& sequence,
                              uint64_t max_sequential_skip_in_iteration,
Y
Yi Wu 已提交
1550
                              uint64_t version_number,
1551 1552
                              ReadCallback* read_callback, DBImpl* db_impl,
                              ColumnFamilyData* cfd, bool allow_blob,
1553
                              bool allow_refresh) {
S
Siying Dong 已提交
1554
  auto mem = arena_.AllocateAligned(sizeof(DBIter));
1555 1556 1557 1558
  db_iter_ = new (mem) DBIter(env, read_options, cf_options, mutable_cf_options,
                              cf_options.user_comparator, nullptr, sequence,
                              true, max_sequential_skip_in_iteration,
                              read_callback, db_impl, cfd, allow_blob);
S
Siying Dong 已提交
1559
  sv_number_ = version_number;
1560
  allow_refresh_ = allow_refresh;
S
Siying Dong 已提交
1561 1562 1563
}

Status ArenaWrappedDBIter::Refresh() {
1564
  if (cfd_ == nullptr || db_impl_ == nullptr || !allow_refresh_) {
S
Siying Dong 已提交
1565 1566 1567
    return Status::NotSupported("Creating renew iterator is not allowed.");
  }
  assert(db_iter_ != nullptr);
1568 1569 1570
  // TODO(yiwu): For last_seq_same_as_publish_seq_==false, this is not the
  // correct behavior. Will be corrected automatically when we take a snapshot
  // here for the case of WritePreparedTxnDB.
S
Siying Dong 已提交
1571 1572 1573 1574 1575 1576 1577 1578 1579
  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());
1580 1581 1582
    if (read_callback_) {
      read_callback_->Refresh(latest_seq);
    }
1583 1584
    Init(env, read_options_, *(cfd_->ioptions()), sv->mutable_cf_options,
         latest_seq, sv->mutable_cf_options.max_sequential_skip_in_iterations,
1585 1586
         cur_sv_number, read_callback_, db_impl_, cfd_, allow_blob_,
         allow_refresh_);
S
Siying Dong 已提交
1587 1588

    InternalIterator* internal_iter = db_impl_->NewInternalIterator(
1589 1590
        read_options_, cfd_, sv, &arena_, db_iter_->GetRangeDelAggregator(),
        latest_seq);
S
Siying Dong 已提交
1591 1592 1593 1594 1595 1596
    SetIterUnderDBIter(internal_iter);
  } else {
    db_iter_->set_sequence(latest_seq);
    db_iter_->set_valid(false);
  }
  return Status::OK();
1597
}
J
jorlow@chromium.org 已提交
1598

1599
ArenaWrappedDBIter* NewArenaWrappedDbIterator(
1600
    Env* env, const ReadOptions& read_options,
1601 1602
    const ImmutableCFOptions& cf_options,
    const MutableCFOptions& mutable_cf_options, const SequenceNumber& sequence,
S
Siying Dong 已提交
1603
    uint64_t max_sequential_skip_in_iterations, uint64_t version_number,
Y
Yi Wu 已提交
1604
    ReadCallback* read_callback, DBImpl* db_impl, ColumnFamilyData* cfd,
1605
    bool allow_blob, bool allow_refresh) {
1606
  ArenaWrappedDBIter* iter = new ArenaWrappedDBIter();
1607
  iter->Init(env, read_options, cf_options, mutable_cf_options, sequence,
Y
Yi Wu 已提交
1608
             max_sequential_skip_in_iterations, version_number, read_callback,
1609
             db_impl, cfd, allow_blob, allow_refresh);
1610
  if (db_impl != nullptr && cfd != nullptr && allow_refresh) {
Y
Yi Wu 已提交
1611 1612
    iter->StoreRefreshInfo(read_options, db_impl, cfd, read_callback,
                           allow_blob);
S
Siying Dong 已提交
1613
  }
1614

1615
  return iter;
J
jorlow@chromium.org 已提交
1616 1617
}

1618
}  // namespace rocksdb