db_iter.cc 50.1 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 "monitoring/perf_context_imp.h"
20 21 22
#include "rocksdb/env.h"
#include "rocksdb/iterator.h"
#include "rocksdb/merge_operator.h"
23
#include "rocksdb/options.h"
S
sdong 已提交
24
#include "table/internal_iterator.h"
25
#include "util/arena.h"
26
#include "util/filename.h"
J
jorlow@chromium.org 已提交
27 28
#include "util/logging.h"
#include "util/mutexlock.h"
29
#include "util/string_util.h"
J
jorlow@chromium.org 已提交
30

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

#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 已提交
51
class DBIter final: public Iterator {
J
jorlow@chromium.org 已提交
52
 public:
53
  // The following is grossly complicated. TODO: clean it up
J
jorlow@chromium.org 已提交
54 55 56 57 58 59 60 61 62 63
  // 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
  };

64 65 66 67 68 69 70 71 72 73 74 75 76 77 78
  // 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;
79
      skip_count_ = 0;
80 81 82 83 84 85 86 87
    }

    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_);
88
      RecordTick(global_statistics, NUMBER_ITER_SKIP, skip_count_);
89
      PERF_COUNTER_ADD(iter_read_bytes, bytes_read_);
90 91 92 93 94 95 96 97 98 99 100 101 102
      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_;
103 104
    // Map to Tickers::NUMBER_ITER_SKIP
    uint64_t skip_count_;
105 106
  };

S
Siying Dong 已提交
107
  DBIter(Env* _env, const ReadOptions& read_options,
108
         const ImmutableCFOptions& cf_options, const Comparator* cmp,
S
sdong 已提交
109
         InternalIterator* iter, SequenceNumber s, bool arena_mode,
Y
Yi Wu 已提交
110 111
         uint64_t max_sequential_skip_in_iterations,
         ReadCallback* read_callback, bool allow_blob)
112
      : arena_mode_(arena_mode),
S
Siying Dong 已提交
113
        env_(_env),
114
        logger_(cf_options.info_log),
J
jorlow@chromium.org 已提交
115
        user_comparator_(cmp),
116
        merge_operator_(cf_options.merge_operator),
J
jorlow@chromium.org 已提交
117 118
        iter_(iter),
        sequence_(s),
J
jorlow@chromium.org 已提交
119
        direction_(kForward),
120
        valid_(false),
121
        current_entry_is_merged_(false),
122
        statistics_(cf_options.statistics),
123
        num_internal_keys_skipped_(0),
124
        iterate_lower_bound_(read_options.iterate_lower_bound),
125 126 127 128 129
        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,
Y
Yi Wu 已提交
130
                       true /* collapse_deletions */),
Y
Yi Wu 已提交
131
        read_callback_(read_callback),
132
        allow_blob_(allow_blob),
133 134
        is_blob_(false),
        start_seqnum_(read_options.iter_start_seqnum) {
L
Lei Jin 已提交
135
    RecordTick(statistics_, NO_ITERATORS);
136
    prefix_extractor_ = cf_options.prefix_extractor;
137
    max_skip_ = max_sequential_skip_in_iterations;
138
    max_skippable_internal_keys_ = read_options.max_skippable_internal_keys;
139 140 141 142 143 144
    if (pin_thru_lifetime_) {
      pinned_iters_mgr_.StartPinning();
    }
    if (iter_) {
      iter_->SetPinnedItersMgr(&pinned_iters_mgr_);
    }
J
jorlow@chromium.org 已提交
145 146
  }
  virtual ~DBIter() {
147
    // Release pinned data if any
148 149 150
    if (pinned_iters_mgr_.PinningEnabled()) {
      pinned_iters_mgr_.ReleasePinnedData();
    }
151 152 153
    // Compiler warning issue filed:
    // https://github.com/facebook/rocksdb/issues/3013
    RecordTick(statistics_, NO_ITERATORS, uint64_t(-1));
154
    ResetInternalKeysSkippedCounter();
155
    local_stats_.BumpGlobalStatistics(statistics_);
156 157 158
    if (!arena_mode_) {
      delete iter_;
    } else {
S
sdong 已提交
159
      iter_->~InternalIterator();
160 161
    }
  }
S
sdong 已提交
162
  virtual void SetIter(InternalIterator* iter) {
163 164
    assert(iter_ == nullptr);
    iter_ = iter;
165
    iter_->SetPinnedItersMgr(&pinned_iters_mgr_);
J
jorlow@chromium.org 已提交
166
  }
A
Andrew Kryczka 已提交
167 168 169 170
  virtual RangeDelAggregator* GetRangeDelAggregator() {
    return &range_del_agg_;
  }

I
Igor Sugak 已提交
171 172
  virtual bool Valid() const override { return valid_; }
  virtual Slice key() const override {
J
jorlow@chromium.org 已提交
173
    assert(valid_);
174 175 176 177 178 179
    if(start_seqnum_ > 0) {
      return saved_key_.GetInternalKey();
    } else {
      return saved_key_.GetUserKey();
    }

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

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

I
Igor Sugak 已提交
227 228 229
  virtual void Next() override;
  virtual void Prev() override;
  virtual void Seek(const Slice& target) override;
A
Aaron Gao 已提交
230
  virtual void SeekForPrev(const Slice& target) override;
I
Igor Sugak 已提交
231 232
  virtual void SeekToFirst() override;
  virtual void SeekToLast() override;
S
Siying Dong 已提交
233 234 235
  Env* env() { return env_; }
  void set_sequence(uint64_t s) { sequence_ = s; }
  void set_valid(bool v) { valid_ = v; }
J
jorlow@chromium.org 已提交
236

J
jorlow@chromium.org 已提交
237
 private:
238
  void ReverseToForward();
239
  void ReverseToBackward();
S
Stanislau Hlebik 已提交
240 241 242 243 244 245
  void PrevInternal();
  void FindParseableKey(ParsedInternalKey* ikey, Direction direction);
  bool FindValueForCurrentKey();
  bool FindValueForCurrentKeyUsingSeek();
  void FindPrevUserKey();
  void FindNextUserKey();
246 247
  inline void FindNextUserEntry(bool skipping, bool prefix_check);
  void FindNextUserEntryInternal(bool skipping, bool prefix_check);
J
jorlow@chromium.org 已提交
248
  bool ParseKey(ParsedInternalKey* key);
249
  void MergeValuesNewToOld();
250
  bool TooManyInternalKeysSkipped(bool increment = true);
Y
Yi Wu 已提交
251
  bool IsVisible(SequenceNumber sequence);
J
jorlow@chromium.org 已提交
252

253 254 255 256 257 258 259 260 261 262
  // 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() {
263 264
    if (!pin_thru_lifetime_ && pinned_iters_mgr_.PinningEnabled()) {
      pinned_iters_mgr_.ReleasePinnedData();
265 266 267
    }
  }

J
jorlow@chromium.org 已提交
268 269 270 271 272 273 274 275 276
  inline void ClearSavedValue() {
    if (saved_value_.capacity() > 1048576) {
      std::string empty;
      swap(empty, saved_value_);
    } else {
      saved_value_.clear();
    }
  }

277
  inline void ResetInternalKeysSkippedCounter() {
278 279 280 281
    local_stats_.skip_count_ += num_internal_keys_skipped_;
    if (valid_) {
      local_stats_.skip_count_--;
    }
282 283 284
    num_internal_keys_skipped_ = 0;
  }

285
  const SliceTransform* prefix_extractor_;
286
  bool arena_mode_;
J
jorlow@chromium.org 已提交
287
  Env* const env_;
I
Igor Canadi 已提交
288
  Logger* logger_;
J
jorlow@chromium.org 已提交
289
  const Comparator* const user_comparator_;
290
  const MergeOperator* const merge_operator_;
S
sdong 已提交
291
  InternalIterator* iter_;
S
Siying Dong 已提交
292
  SequenceNumber sequence_;
J
jorlow@chromium.org 已提交
293

J
jorlow@chromium.org 已提交
294
  Status status_;
S
Stanislau Hlebik 已提交
295
  IterKey saved_key_;
296 297 298 299
  // 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 已提交
300
  std::string saved_value_;
301
  Slice pinned_value_;
J
jorlow@chromium.org 已提交
302
  Direction direction_;
J
jorlow@chromium.org 已提交
303
  bool valid_;
304
  bool current_entry_is_merged_;
305
  // for prefix seek mode to support prev()
306
  Statistics* statistics_;
307
  uint64_t max_skip_;
308 309
  uint64_t max_skippable_internal_keys_;
  uint64_t num_internal_keys_skipped_;
310
  const Slice* iterate_lower_bound_;
311
  const Slice* iterate_upper_bound_;
312 313 314
  IterKey prefix_start_buf_;
  Slice prefix_start_key_;
  const bool prefix_same_as_start_;
315 316 317
  // 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_;
318
  const bool total_order_seek_;
319
  // List of operands for merge operator.
320
  MergeContext merge_context_;
A
Andrew Kryczka 已提交
321
  RangeDelAggregator range_del_agg_;
322
  LocalStatistics local_stats_;
323
  PinnedIteratorsManager pinned_iters_mgr_;
Y
Yi Wu 已提交
324
  ReadCallback* read_callback_;
Y
Yi Wu 已提交
325 326
  bool allow_blob_;
  bool is_blob_;
327 328 329
  // 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 已提交
330 331 332 333 334 335 336 337 338

  // 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");
339 340
    ROCKS_LOG_ERROR(logger_, "corrupted internal key in DBIter: %s",
                    iter_->key().ToString(true).c_str());
J
jorlow@chromium.org 已提交
341 342 343 344 345 346
    return false;
  } else {
    return true;
  }
}

J
jorlow@chromium.org 已提交
347 348 349
void DBIter::Next() {
  assert(valid_);

350 351
  // Release temporarily pinned blocks from last operation
  ReleaseTempPinnedData();
352
  ResetInternalKeysSkippedCounter();
S
Stanislau Hlebik 已提交
353
  if (direction_ == kReverse) {
354
    ReverseToForward();
355 356 357 358 359 360 361
  } 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();
362
    PERF_COUNTER_ADD(internal_key_skipped_count, 1);
J
jorlow@chromium.org 已提交
363
  }
J
jorlow@chromium.org 已提交
364

365 366 367
  if (statistics_ != nullptr) {
    local_stats_.next_count_++;
  }
368 369
  // Now we point to the next internal position, for both of merge and
  // not merge cases.
370 371 372 373
  if (!iter_->Valid()) {
    valid_ = false;
    return;
  }
374
  FindNextUserEntry(true /* skipping the current user key */, prefix_same_as_start_);
375 376 377 378
  if (statistics_ != nullptr && valid_) {
    local_stats_.next_found_count_++;
    local_stats_.bytes_read_ += (key().size() + value().size());
  }
J
jorlow@chromium.org 已提交
379 380
}

381 382 383 384 385 386 387
// 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
388
//       a delete marker or a sequence number higher than sequence_
389
//       saved_key_ MUST have a proper user_key before calling this function
390 391 392 393 394 395
//
// 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) {
396
  PERF_TIMER_GUARD(find_next_user_entry_time);
397
  FindNextUserEntryInternal(skipping, prefix_check);
398 399 400
}

// Actual implementation of DBIter::FindNextUserEntry()
401
void DBIter::FindNextUserEntryInternal(bool skipping, bool prefix_check) {
J
jorlow@chromium.org 已提交
402 403 404
  // Loop until we hit an acceptable entry to yield
  assert(iter_->Valid());
  assert(direction_ == kForward);
405
  current_entry_is_merged_ = false;
406 407 408 409 410 411 412 413 414 415 416

  // 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.
417
  uint64_t num_skipped = 0;
418

Y
Yi Wu 已提交
419 420
  is_blob_ = false;

J
jorlow@chromium.org 已提交
421
  do {
422
    if (!ParseKey(&ikey_)) {
423 424 425 426
      // Skip corrupted keys.
      iter_->Next();
      continue;
    }
427

428
    if (iterate_upper_bound_ != nullptr &&
429
        user_comparator_->Compare(ikey_.user_key, *iterate_upper_bound_) >= 0) {
430 431
      break;
    }
432

433
    if (prefix_extractor_ && prefix_check &&
434 435
        prefix_extractor_->Transform(ikey_.user_key)
                .compare(prefix_start_key_) != 0) {
436 437 438
      break;
    }

439 440 441 442
    if (TooManyInternalKeysSkipped()) {
      return;
    }

Y
Yi Wu 已提交
443
    if (IsVisible(ikey_.sequence)) {
444 445
      if (skipping && user_comparator_->Compare(ikey_.user_key,
                                                saved_key_.GetUserKey()) <= 0) {
446 447 448 449
        num_skipped++;  // skip this entry
        PERF_COUNTER_ADD(internal_key_skipped_count, 1);
      } else {
        num_skipped = 0;
450
        switch (ikey_.type) {
451 452 453 454
          case kTypeDeletion:
          case kTypeSingleDeletion:
            // Arrange to skip all upcoming entries for this key since
            // they are hidden by this deletion.
455 456 457 458 459 460 461 462 463 464 465
            // if iterartor specified start_seqnum we
            // 1) return internal key, including the type
            // 2) return ikey only if ikey.seqnum >= start_seqnum_
            // not that if deletion seqnum is < start_seqnum_ we
            // just skip it like in normal iterator.
            if (start_seqnum_ > 0 && ikey_.sequence >= start_seqnum_)  {
              saved_key_.SetInternalKey(ikey_);
              valid_=true;
              return;
            } else {
              saved_key_.SetUserKey(
466 467
                ikey_.user_key,
                !pin_thru_lifetime_ || !iter_->IsKeyPinned() /* copy */);
468 469 470
                skipping = true;
                PERF_COUNTER_ADD(internal_delete_skipped_count, 1);
            }
471 472
            break;
          case kTypeValue:
Y
Yi Wu 已提交
473
          case kTypeBlobIndex:
474 475 476 477 478 479 480 481
            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 已提交
482
                valid_ = true;
483 484 485 486 487 488 489
                return;
              } else {
                // this key and all previous versions shouldn't be included,
                // skipping
                saved_key_.SetUserKey(ikey_.user_key,
                  !pin_thru_lifetime_ || !iter_->IsKeyPinned() /* copy */);
                skipping = true;
Y
Yi Wu 已提交
490
              }
491
            } else {
492 493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518
              saved_key_.SetUserKey(
                  ikey_.user_key,
                  !pin_thru_lifetime_ || !iter_->IsKeyPinned() /* 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 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;
                } else {
                  is_blob_ = true;
                  valid_ = true;
                }
                return;
              } else {
                valid_ = true;
                return;
              }
519 520 521
            }
            break;
          case kTypeMerge:
522
            saved_key_.SetUserKey(
523 524
                ikey_.user_key,
                !pin_thru_lifetime_ || !iter_->IsKeyPinned() /* copy */);
525
            if (range_del_agg_.ShouldDelete(
526 527
                    ikey_, RangeDelAggregator::RangePositioningMode::
                               kForwardTraversal)) {
528 529 530 531 532 533 534 535 536 537 538 539 540 541 542 543 544
              // 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;
545
        }
J
jorlow@chromium.org 已提交
546
      }
547 548 549 550 551 552
    } 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.
553
      if (user_comparator_->Compare(ikey_.user_key, saved_key_.GetUserKey()) <=
554
          0) {
555 556
        num_skipped++;
      } else {
557
        saved_key_.SetUserKey(
558
            ikey_.user_key,
559
            !iter_->IsKeyPinned() || !pin_thru_lifetime_ /* copy */);
560 561 562
        skipping = false;
        num_skipped = 0;
      }
J
jorlow@chromium.org 已提交
563
    }
564 565 566 567

    // 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_) {
568 569
      num_skipped = 0;
      std::string last_key;
570 571 572 573
      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).
574 575
        AppendInternalKey(&last_key, ParsedInternalKey(saved_key_.GetUserKey(),
                                                       0, kTypeDeletion));
576 577 578 579 580 581 582 583 584
        // 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,
585
                          ParsedInternalKey(saved_key_.GetUserKey(), sequence_,
586 587
                                            kValueTypeForSeek));
      }
588 589 590 591 592
      iter_->Seek(last_key);
      RecordTick(statistics_, NUMBER_OF_RESEEKS_IN_ITERATION);
    } else {
      iter_->Next();
    }
J
jorlow@chromium.org 已提交
593 594
  } while (iter_->Valid());
  valid_ = false;
J
jorlow@chromium.org 已提交
595 596
}

597 598 599 600 601 602 603
// 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() {
604
  if (!merge_operator_) {
605
    ROCKS_LOG_ERROR(logger_, "Options::merge_operator is null.");
606
    status_ = Status::InvalidArgument("merge_operator_ must be set.");
607 608
    valid_ = false;
    return;
D
Deon Nicholas 已提交
609
  }
610

611 612
  // Temporarily pin the blocks that hold merge operands
  TempPinData();
613
  merge_context_.Clear();
614
  // Start the merge process by pushing the first operand
615 616
  merge_context_.PushOperand(iter_->value(),
                             iter_->IsValuePinned() /* operand_pinned */);
617
  TEST_SYNC_POINT("DBIter::MergeValuesNewToOld:PushedFirstOperand");
618 619

  ParsedInternalKey ikey;
620
  Status s;
621
  for (iter_->Next(); iter_->Valid(); iter_->Next()) {
622
    TEST_SYNC_POINT("DBIter::MergeValuesNewToOld:SteppedToNextOperand");
623 624 625 626 627
    if (!ParseKey(&ikey)) {
      // skip corrupted key
      continue;
    }

628
    if (!user_comparator_->Equal(ikey.user_key, saved_key_.GetUserKey())) {
629 630
      // hit the next user key, stop right here
      break;
A
Andrew Kryczka 已提交
631
    } else if (kTypeDeletion == ikey.type || kTypeSingleDeletion == ikey.type ||
632 633 634
               range_del_agg_.ShouldDelete(
                   ikey, RangeDelAggregator::RangePositioningMode::
                             kForwardTraversal)) {
635 636 637 638
      // hit a delete with the same user key, stop right here
      // iter_ is positioned after delete
      iter_->Next();
      break;
A
Andres Noetzli 已提交
639
    } else if (kTypeValue == ikey.type) {
640 641 642
      // 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.
I
Igor Canadi 已提交
643
      const Slice val = iter_->value();
644 645
      s = MergeHelper::TimedFullMerge(
          merge_operator_, ikey.user_key, &val, merge_context_.GetOperands(),
646
          &saved_value_, logger_, statistics_, env_, &pinned_value_, true);
647
      if (!s.ok()) {
Y
Yi Wu 已提交
648
        valid_ = false;
649 650
        status_ = s;
      }
651 652 653
      // iter_ is positioned after put
      iter_->Next();
      return;
A
Andres Noetzli 已提交
654
    } else if (kTypeMerge == ikey.type) {
655 656
      // hit a merge, add the value as an operand and run associative merge.
      // when complete, add result to operands and continue.
657 658
      merge_context_.PushOperand(iter_->value(),
                                 iter_->IsValuePinned() /* operand_pinned */);
659
      PERF_COUNTER_ADD(internal_merge_count, 1);
Y
Yi Wu 已提交
660 661 662 663 664 665 666 667 668 669 670 671
    } 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;
      return;
A
Andres Noetzli 已提交
672 673
    } else {
      assert(false);
674 675 676
    }
  }

677 678 679 680
  // 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.
681 682 683
  s = MergeHelper::TimedFullMerge(merge_operator_, saved_key_.GetUserKey(),
                                  nullptr, merge_context_.GetOperands(),
                                  &saved_value_, logger_, statistics_, env_,
684
                                  &pinned_value_, true);
685
  if (!s.ok()) {
Y
Yi Wu 已提交
686
    valid_ = false;
687 688
    status_ = s;
  }
689 690
}

J
jorlow@chromium.org 已提交
691 692
void DBIter::Prev() {
  assert(valid_);
693
  ReleaseTempPinnedData();
694
  ResetInternalKeysSkippedCounter();
S
Stanislau Hlebik 已提交
695
  if (direction_ == kForward) {
696
    ReverseToBackward();
S
Stanislau Hlebik 已提交
697 698
  }
  PrevInternal();
M
Manuel Ung 已提交
699
  if (statistics_ != nullptr) {
700
    local_stats_.prev_count_++;
M
Manuel Ung 已提交
701
    if (valid_) {
702 703
      local_stats_.prev_found_count_++;
      local_stats_.bytes_read_ += (key().size() + value().size());
M
Manuel Ung 已提交
704 705
    }
  }
S
Stanislau Hlebik 已提交
706
}
J
jorlow@chromium.org 已提交
707

708 709 710 711
void DBIter::ReverseToForward() {
  if (prefix_extractor_ != nullptr && !total_order_seek_) {
    IterKey last_key;
    last_key.SetInternalKey(ParsedInternalKey(
712 713
        saved_key_.GetUserKey(), kMaxSequenceNumber, kValueTypeForSeek));
    iter_->Seek(last_key.GetInternalKey());
714 715 716 717 718
  }
  FindNextUserKey();
  direction_ = kForward;
  if (!iter_->Valid()) {
    iter_->SeekToFirst();
719
    range_del_agg_.InvalidateTombstoneMapPositions();
720 721 722
  }
}

723
void DBIter::ReverseToBackward() {
724 725
  if (prefix_extractor_ != nullptr && !total_order_seek_) {
    IterKey last_key;
726 727 728
    last_key.SetInternalKey(ParsedInternalKey(saved_key_.GetUserKey(), 0,
                                              kValueTypeForSeekForPrev));
    iter_->SeekForPrev(last_key.GetInternalKey());
729
  }
730 731 732 733 734
  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();
735
      range_del_agg_.InvalidateTombstoneMapPositions();
736 737 738 739
    }
    ParsedInternalKey ikey;
    FindParseableKey(&ikey, kReverse);
    while (iter_->Valid() &&
740 741
           user_comparator_->Compare(ikey.user_key, saved_key_.GetUserKey()) >
               0) {
S
Siying Dong 已提交
742
      assert(ikey.sequence != kMaxSequenceNumber);
Y
Yi Wu 已提交
743
      if (!IsVisible(ikey.sequence)) {
744 745 746 747
        PERF_COUNTER_ADD(internal_recent_skipped_count, 1);
      } else {
        PERF_COUNTER_ADD(internal_key_skipped_count, 1);
      }
748 749 750 751 752 753 754 755
      iter_->Prev();
      FindParseableKey(&ikey, kReverse);
    }
  }
#ifndef NDEBUG
  if (iter_->Valid()) {
    ParsedInternalKey ikey;
    assert(ParseKey(&ikey));
756 757
    assert(user_comparator_->Compare(ikey.user_key, saved_key_.GetUserKey()) <=
           0);
758 759 760 761 762 763 764
  }
#endif

  FindPrevUserKey();
  direction_ = kReverse;
}

S
Stanislau Hlebik 已提交
765 766 767 768
void DBIter::PrevInternal() {
  if (!iter_->Valid()) {
    valid_ = false;
    return;
769 770
  }

S
Stanislau Hlebik 已提交
771 772 773
  ParsedInternalKey ikey;

  while (iter_->Valid()) {
774 775 776
    saved_key_.SetUserKey(
        ExtractUserKey(iter_->key()),
        !iter_->IsKeyPinned() || !pin_thru_lifetime_ /* copy */);
777

778 779 780 781 782 783 784 785
    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;
    }

786 787 788 789 790 791 792 793
    if (iterate_lower_bound_ != nullptr &&
        user_comparator_->Compare(saved_key_.GetUserKey(),
                                  *iterate_lower_bound_) < 0) {
      // We've iterated earlier than the user-specified lower bound.
      valid_ = false;
      return;
    }

S
Stanislau Hlebik 已提交
794
    if (FindValueForCurrentKey()) {
J
jorlow@chromium.org 已提交
795 796 797
      if (!iter_->Valid()) {
        return;
      }
S
Stanislau Hlebik 已提交
798
      FindParseableKey(&ikey, kReverse);
799
      if (user_comparator_->Equal(ikey.user_key, saved_key_.GetUserKey())) {
S
Stanislau Hlebik 已提交
800
        FindPrevUserKey();
J
jorlow@chromium.org 已提交
801
      }
S
Stanislau Hlebik 已提交
802
      return;
J
jorlow@chromium.org 已提交
803
    }
804 805 806 807 808

    if (TooManyInternalKeysSkipped(false)) {
      return;
    }

S
Stanislau Hlebik 已提交
809 810 811 812
    if (!iter_->Valid()) {
      break;
    }
    FindParseableKey(&ikey, kReverse);
813
    if (user_comparator_->Equal(ikey.user_key, saved_key_.GetUserKey())) {
S
Stanislau Hlebik 已提交
814 815 816 817
      FindPrevUserKey();
    }
  }
  // We haven't found any key - iterator is not valid
A
Aaron Gao 已提交
818
  // Or the prefix is different than start prefix
819
  assert(!iter_->Valid());
S
Stanislau Hlebik 已提交
820
  valid_ = false;
J
jorlow@chromium.org 已提交
821 822
}

S
Stanislau Hlebik 已提交
823
// This function checks, if the entry with biggest sequence_number <= sequence_
A
Andres Noetzli 已提交
824 825
// is non kTypeDeletion or kTypeSingleDeletion. If it's not, we save value in
// saved_value_
S
Stanislau Hlebik 已提交
826 827
bool DBIter::FindValueForCurrentKey() {
  assert(iter_->Valid());
828
  merge_context_.Clear();
829
  current_entry_is_merged_ = false;
A
Andres Noetzli 已提交
830 831
  // last entry before merge (could be kTypeDeletion, kTypeSingleDeletion or
  // kTypeValue)
S
Stanislau Hlebik 已提交
832 833
  ValueType last_not_merge_type = kTypeDeletion;
  ValueType last_key_entry_type = kTypeDeletion;
J
jorlow@chromium.org 已提交
834

S
Stanislau Hlebik 已提交
835 836 837
  ParsedInternalKey ikey;
  FindParseableKey(&ikey, kReverse);

838 839 840
  // Temporarily pin blocks that hold (merge operands / the value)
  ReleaseTempPinnedData();
  TempPinData();
S
Stanislau Hlebik 已提交
841
  size_t num_skipped = 0;
Y
Yi Wu 已提交
842
  while (iter_->Valid() && IsVisible(ikey.sequence) &&
843
         user_comparator_->Equal(ikey.user_key, saved_key_.GetUserKey())) {
844 845 846 847
    if (TooManyInternalKeysSkipped()) {
      return false;
    }

S
Stanislau Hlebik 已提交
848 849 850 851 852 853 854 855
    // 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:
Y
Yi Wu 已提交
856
      case kTypeBlobIndex:
857 858 859
        if (range_del_agg_.ShouldDelete(
                ikey,
                RangeDelAggregator::RangePositioningMode::kBackwardTraversal)) {
A
Andrew Kryczka 已提交
860 861 862 863 864 865
          last_key_entry_type = kTypeRangeDeletion;
          PERF_COUNTER_ADD(internal_delete_skipped_count, 1);
        } else {
          assert(iter_->IsValuePinned());
          pinned_value_ = iter_->value();
        }
866
        merge_context_.Clear();
A
Andrew Kryczka 已提交
867
        last_not_merge_type = last_key_entry_type;
S
Stanislau Hlebik 已提交
868 869
        break;
      case kTypeDeletion:
A
Andres Noetzli 已提交
870
      case kTypeSingleDeletion:
871
        merge_context_.Clear();
A
Andres Noetzli 已提交
872
        last_not_merge_type = last_key_entry_type;
873
        PERF_COUNTER_ADD(internal_delete_skipped_count, 1);
S
Stanislau Hlebik 已提交
874 875
        break;
      case kTypeMerge:
876 877 878
        if (range_del_agg_.ShouldDelete(
                ikey,
                RangeDelAggregator::RangePositioningMode::kBackwardTraversal)) {
A
Andrew Kryczka 已提交
879 880 881 882 883 884 885 886
          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 */);
887
          PERF_COUNTER_ADD(internal_merge_count, 1);
A
Andrew Kryczka 已提交
888
        }
S
Stanislau Hlebik 已提交
889 890 891 892 893
        break;
      default:
        assert(false);
    }

894
    PERF_COUNTER_ADD(internal_key_skipped_count, 1);
895
    assert(user_comparator_->Equal(ikey.user_key, saved_key_.GetUserKey()));
S
Stanislau Hlebik 已提交
896 897 898 899 900
    iter_->Prev();
    ++num_skipped;
    FindParseableKey(&ikey, kReverse);
  }

901
  Status s;
Y
Yi Wu 已提交
902
  is_blob_ = false;
S
Stanislau Hlebik 已提交
903 904
  switch (last_key_entry_type) {
    case kTypeDeletion:
A
Andres Noetzli 已提交
905
    case kTypeSingleDeletion:
A
Andrew Kryczka 已提交
906
    case kTypeRangeDeletion:
S
Stanislau Hlebik 已提交
907 908 909
      valid_ = false;
      return false;
    case kTypeMerge:
910
      current_entry_is_merged_ = true;
A
Aaron Gao 已提交
911
      if (last_not_merge_type == kTypeDeletion ||
A
Andrew Kryczka 已提交
912 913
          last_not_merge_type == kTypeSingleDeletion ||
          last_not_merge_type == kTypeRangeDeletion) {
914 915 916
        s = MergeHelper::TimedFullMerge(
            merge_operator_, saved_key_.GetUserKey(), nullptr,
            merge_context_.GetOperands(), &saved_value_, logger_, statistics_,
917
            env_, &pinned_value_, true);
Y
Yi Wu 已提交
918 919 920 921 922 923 924 925 926 927 928 929
      } 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;
        return true;
930
      } else {
S
Stanislau Hlebik 已提交
931
        assert(last_not_merge_type == kTypeValue);
932
        s = MergeHelper::TimedFullMerge(
933
            merge_operator_, saved_key_.GetUserKey(), &pinned_value_,
934
            merge_context_.GetOperands(), &saved_value_, logger_, statistics_,
935
            env_, &pinned_value_, true);
936
      }
S
Stanislau Hlebik 已提交
937 938 939 940
      break;
    case kTypeValue:
      // do nothing - we've already has value in saved_value_
      break;
Y
Yi Wu 已提交
941 942 943 944 945 946 947 948 949 950 951
    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;
        return true;
      }
      is_blob_ = true;
      break;
S
Stanislau Hlebik 已提交
952 953 954
    default:
      assert(false);
      break;
J
jorlow@chromium.org 已提交
955
  }
Y
Yi Wu 已提交
956 957 958 959
  if (s.ok()) {
    valid_ = true;
  } else {
    valid_ = false;
960 961
    status_ = s;
  }
S
Stanislau Hlebik 已提交
962 963
  return true;
}
J
jorlow@chromium.org 已提交
964

S
Stanislau Hlebik 已提交
965 966 967
// This function is used in FindValueForCurrentKey.
// We use Seek() function instead of Prev() to find necessary value
bool DBIter::FindValueForCurrentKeyUsingSeek() {
968 969 970
  // FindValueForCurrentKey will enable pinning before calling
  // FindValueForCurrentKeyUsingSeek()
  assert(pinned_iters_mgr_.PinningEnabled());
S
Stanislau Hlebik 已提交
971
  std::string last_key;
972 973
  AppendInternalKey(&last_key, ParsedInternalKey(saved_key_.GetUserKey(),
                                                 sequence_, kValueTypeForSeek));
S
Stanislau Hlebik 已提交
974 975 976 977 978 979
  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);
980 981 982 983 984 985 986 987 988 989 990
  assert(iter_->Valid());
  assert(user_comparator_->Equal(ikey.user_key, saved_key_.GetUserKey()));

  // In case read_callback presents, the value we seek to may not be visible.
  // Seek for the next value that's visible.
  while (!IsVisible(ikey.sequence)) {
    iter_->Next();
    FindParseableKey(&ikey, kForward);
    assert(iter_->Valid());
    assert(user_comparator_->Equal(ikey.user_key, saved_key_.GetUserKey()));
  }
S
Stanislau Hlebik 已提交
991

A
Andrew Kryczka 已提交
992
  if (ikey.type == kTypeDeletion || ikey.type == kTypeSingleDeletion ||
993 994
      range_del_agg_.ShouldDelete(
          ikey, RangeDelAggregator::RangePositioningMode::kBackwardTraversal)) {
J
jorlow@chromium.org 已提交
995
    valid_ = false;
S
Stanislau Hlebik 已提交
996 997
    return false;
  }
Y
Yi Wu 已提交
998 999 1000 1001 1002 1003 1004 1005 1006
  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;
    return true;
  }
  if (ikey.type == kTypeValue || ikey.type == kTypeBlobIndex) {
A
Andrew Kryczka 已提交
1007 1008 1009 1010 1011
    assert(iter_->IsValuePinned());
    pinned_value_ = iter_->value();
    valid_ = true;
    return true;
  }
S
Stanislau Hlebik 已提交
1012 1013 1014

  // kTypeMerge. We need to collect all kTypeMerge values and save them
  // in operands
1015
  current_entry_is_merged_ = true;
1016
  merge_context_.Clear();
1017 1018
  while (
      iter_->Valid() &&
1019
      user_comparator_->Equal(ikey.user_key, saved_key_.GetUserKey()) &&
1020 1021 1022
      ikey.type == kTypeMerge &&
      !range_del_agg_.ShouldDelete(
          ikey, RangeDelAggregator::RangePositioningMode::kBackwardTraversal)) {
1023 1024
    merge_context_.PushOperand(iter_->value(),
                               iter_->IsValuePinned() /* operand_pinned */);
1025
    PERF_COUNTER_ADD(internal_merge_count, 1);
S
Stanislau Hlebik 已提交
1026 1027 1028 1029
    iter_->Next();
    FindParseableKey(&ikey, kForward);
  }

1030
  Status s;
S
Stanislau Hlebik 已提交
1031
  if (!iter_->Valid() ||
1032
      !user_comparator_->Equal(ikey.user_key, saved_key_.GetUserKey()) ||
A
Andrew Kryczka 已提交
1033
      ikey.type == kTypeDeletion || ikey.type == kTypeSingleDeletion ||
1034 1035
      range_del_agg_.ShouldDelete(
          ikey, RangeDelAggregator::RangePositioningMode::kBackwardTraversal)) {
1036
    s = MergeHelper::TimedFullMerge(merge_operator_, saved_key_.GetUserKey(),
1037 1038
                                    nullptr, merge_context_.GetOperands(),
                                    &saved_value_, logger_, statistics_, env_,
1039
                                    &pinned_value_, true);
S
Stanislau Hlebik 已提交
1040 1041
    // Make iter_ valid and point to saved_key_
    if (!iter_->Valid() ||
1042
        !user_comparator_->Equal(ikey.user_key, saved_key_.GetUserKey())) {
S
Stanislau Hlebik 已提交
1043 1044 1045
      iter_->Seek(last_key);
      RecordTick(statistics_, NUMBER_OF_RESEEKS_IN_ITERATION);
    }
Y
Yi Wu 已提交
1046 1047 1048 1049
    if (s.ok()) {
      valid_ = true;
    } else {
      valid_ = false;
1050 1051
      status_ = s;
    }
S
Stanislau Hlebik 已提交
1052 1053 1054
    return true;
  }

I
Igor Canadi 已提交
1055
  const Slice& val = iter_->value();
1056 1057 1058
  s = MergeHelper::TimedFullMerge(merge_operator_, saved_key_.GetUserKey(),
                                  &val, merge_context_.GetOperands(),
                                  &saved_value_, logger_, statistics_, env_,
1059
                                  &pinned_value_, true);
Y
Yi Wu 已提交
1060 1061 1062 1063
  if (s.ok()) {
    valid_ = true;
  } else {
    valid_ = false;
1064 1065
    status_ = s;
  }
S
Stanislau Hlebik 已提交
1066 1067 1068 1069 1070 1071 1072 1073 1074 1075 1076 1077 1078 1079
  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() &&
1080
         !user_comparator_->Equal(ikey.user_key, saved_key_.GetUserKey())) {
S
Stanislau Hlebik 已提交
1081 1082 1083 1084 1085 1086 1087 1088 1089 1090 1091 1092 1093
    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);
1094
  int cmp;
1095 1096 1097
  while (iter_->Valid() &&
         ((cmp = user_comparator_->Compare(ikey.user_key,
                                           saved_key_.GetUserKey())) == 0 ||
Y
Yi Wu 已提交
1098
          (cmp > 0 && !IsVisible(ikey.sequence)))) {
1099 1100 1101 1102
    if (TooManyInternalKeysSkipped()) {
      return;
    }

1103 1104 1105 1106 1107
    if (cmp == 0) {
      if (num_skipped >= max_skip_) {
        num_skipped = 0;
        IterKey last_key;
        last_key.SetInternalKey(ParsedInternalKey(
1108 1109
            saved_key_.GetUserKey(), kMaxSequenceNumber, kValueTypeForSeek));
        iter_->Seek(last_key.GetInternalKey());
1110 1111 1112 1113
        RecordTick(statistics_, NUMBER_OF_RESEEKS_IN_ITERATION);
      } else {
        ++num_skipped;
      }
S
Stanislau Hlebik 已提交
1114
    }
S
Siying Dong 已提交
1115
    assert(ikey.sequence != kMaxSequenceNumber);
Y
Yi Wu 已提交
1116
    if (!IsVisible(ikey.sequence)) {
1117 1118 1119 1120
      PERF_COUNTER_ADD(internal_recent_skipped_count, 1);
    } else {
      PERF_COUNTER_ADD(internal_key_skipped_count, 1);
    }
S
Stanislau Hlebik 已提交
1121 1122 1123 1124 1125
    iter_->Prev();
    FindParseableKey(&ikey, kReverse);
  }
}

1126 1127 1128 1129 1130 1131 1132 1133 1134 1135 1136 1137
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 已提交
1138 1139 1140 1141 1142
bool DBIter::IsVisible(SequenceNumber sequence) {
  return sequence <= sequence_ &&
         (read_callback_ == nullptr || read_callback_->IsCommitted(sequence));
}

S
Stanislau Hlebik 已提交
1143 1144 1145 1146 1147 1148 1149 1150
// Skip all unparseable keys
void DBIter::FindParseableKey(ParsedInternalKey* ikey, Direction direction) {
  while (iter_->Valid() && !ParseKey(ikey)) {
    if (direction == kReverse) {
      iter_->Prev();
    } else {
      iter_->Next();
    }
J
jorlow@chromium.org 已提交
1151 1152
  }
}
J
jorlow@chromium.org 已提交
1153

J
jorlow@chromium.org 已提交
1154
void DBIter::Seek(const Slice& target) {
L
Lei Jin 已提交
1155
  StopWatch sw(env_, statistics_, DB_SEEK);
1156
  ReleaseTempPinnedData();
1157
  ResetInternalKeysSkippedCounter();
1158 1159
  saved_key_.Clear();
  saved_key_.SetInternalKey(target, sequence_);
1160

Z
zhangjinpeng1987 已提交
1161 1162 1163 1164 1165 1166 1167
  if (iterate_lower_bound_ != nullptr &&
      user_comparator_->Compare(saved_key_.GetUserKey(),
                                *iterate_lower_bound_) < 0) {
    saved_key_.Clear();
    saved_key_.SetInternalKey(*iterate_lower_bound_, sequence_);
  }

1168 1169
  {
    PERF_TIMER_GUARD(seek_internal_seek_time);
1170
    iter_->Seek(saved_key_.GetInternalKey());
1171
    range_del_agg_.InvalidateTombstoneMapPositions();
1172
  }
M
Manuel Ung 已提交
1173
  RecordTick(statistics_, NUMBER_DB_SEEK);
J
jorlow@chromium.org 已提交
1174
  if (iter_->Valid()) {
1175 1176 1177
    if (prefix_extractor_ && prefix_same_as_start_) {
      prefix_start_key_ = prefix_extractor_->Transform(target);
    }
1178 1179
    direction_ = kForward;
    ClearSavedValue();
1180 1181 1182 1183
    FindNextUserEntry(false /* not skipping */, prefix_same_as_start_);
    if (!valid_) {
      prefix_start_key_.clear();
    }
M
Manuel Ung 已提交
1184 1185
    if (statistics_ != nullptr) {
      if (valid_) {
1186
        // Decrement since we don't want to count this key as skipped
M
Manuel Ung 已提交
1187 1188
        RecordTick(statistics_, NUMBER_DB_SEEK_FOUND);
        RecordTick(statistics_, ITER_BYTES_READ, key().size() + value().size());
1189
        PERF_COUNTER_ADD(iter_read_bytes, key().size() + value().size());
M
Manuel Ung 已提交
1190 1191
      }
    }
J
jorlow@chromium.org 已提交
1192 1193 1194
  } else {
    valid_ = false;
  }
1195

1196
  if (valid_ && prefix_extractor_ && prefix_same_as_start_) {
1197 1198
    prefix_start_buf_.SetUserKey(prefix_start_key_);
    prefix_start_key_ = prefix_start_buf_.GetUserKey();
1199
  }
J
jorlow@chromium.org 已提交
1200
}
J
jorlow@chromium.org 已提交
1201

A
Aaron Gao 已提交
1202 1203 1204
void DBIter::SeekForPrev(const Slice& target) {
  StopWatch sw(env_, statistics_, DB_SEEK);
  ReleaseTempPinnedData();
1205
  ResetInternalKeysSkippedCounter();
A
Aaron Gao 已提交
1206 1207 1208 1209 1210
  saved_key_.Clear();
  // now saved_key is used to store internal key.
  saved_key_.SetInternalKey(target, 0 /* sequence_number */,
                            kValueTypeForSeekForPrev);

Z
zhangjinpeng1987 已提交
1211 1212 1213 1214 1215 1216 1217
  if (iterate_upper_bound_ != nullptr &&
      user_comparator_->Compare(saved_key_.GetUserKey(),
                                *iterate_upper_bound_) >= 0) {
    saved_key_.Clear();
    saved_key_.SetInternalKey(*iterate_upper_bound_, kMaxSequenceNumber);
  }

A
Aaron Gao 已提交
1218 1219
  {
    PERF_TIMER_GUARD(seek_internal_seek_time);
1220
    iter_->SeekForPrev(saved_key_.GetInternalKey());
1221
    range_del_agg_.InvalidateTombstoneMapPositions();
A
Aaron Gao 已提交
1222 1223 1224 1225 1226 1227 1228 1229 1230 1231 1232 1233 1234 1235 1236 1237 1238
  }

  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());
1239
        PERF_COUNTER_ADD(iter_read_bytes, key().size() + value().size());
A
Aaron Gao 已提交
1240 1241 1242 1243 1244 1245
      }
    }
  } else {
    valid_ = false;
  }
  if (valid_ && prefix_extractor_ && prefix_same_as_start_) {
1246 1247
    prefix_start_buf_.SetUserKey(prefix_start_key_);
    prefix_start_key_ = prefix_start_buf_.GetUserKey();
A
Aaron Gao 已提交
1248 1249 1250
  }
}

J
jorlow@chromium.org 已提交
1251
void DBIter::SeekToFirst() {
S
Stanislau Hlebik 已提交
1252
  // Don't use iter_::Seek() if we set a prefix extractor
1253
  // because prefix seek will be used.
1254
  if (prefix_extractor_ != nullptr) {
S
Stanislau Hlebik 已提交
1255 1256
    max_skip_ = std::numeric_limits<uint64_t>::max();
  }
1257 1258 1259 1260
  if (iterate_lower_bound_ != nullptr) {
    Seek(*iterate_lower_bound_);
    return;
  }
J
jorlow@chromium.org 已提交
1261
  direction_ = kForward;
1262
  ReleaseTempPinnedData();
1263
  ResetInternalKeysSkippedCounter();
J
jorlow@chromium.org 已提交
1264
  ClearSavedValue();
1265 1266 1267 1268

  {
    PERF_TIMER_GUARD(seek_internal_seek_time);
    iter_->SeekToFirst();
1269
    range_del_agg_.InvalidateTombstoneMapPositions();
1270 1271
  }

M
Manuel Ung 已提交
1272
  RecordTick(statistics_, NUMBER_DB_SEEK);
J
jorlow@chromium.org 已提交
1273
  if (iter_->Valid()) {
1274 1275 1276
    saved_key_.SetUserKey(
        ExtractUserKey(iter_->key()),
        !iter_->IsKeyPinned() || !pin_thru_lifetime_ /* copy */);
1277
    FindNextUserEntry(false /* not skipping */, false /* no prefix check */);
M
Manuel Ung 已提交
1278 1279 1280 1281
    if (statistics_ != nullptr) {
      if (valid_) {
        RecordTick(statistics_, NUMBER_DB_SEEK_FOUND);
        RecordTick(statistics_, ITER_BYTES_READ, key().size() + value().size());
1282
        PERF_COUNTER_ADD(iter_read_bytes, key().size() + value().size());
M
Manuel Ung 已提交
1283 1284
      }
    }
J
jorlow@chromium.org 已提交
1285 1286
  } else {
    valid_ = false;
J
jorlow@chromium.org 已提交
1287
  }
1288
  if (valid_ && prefix_extractor_ && prefix_same_as_start_) {
1289 1290 1291
    prefix_start_buf_.SetUserKey(
        prefix_extractor_->Transform(saved_key_.GetUserKey()));
    prefix_start_key_ = prefix_start_buf_.GetUserKey();
1292
  }
J
jorlow@chromium.org 已提交
1293 1294
}

J
jorlow@chromium.org 已提交
1295
void DBIter::SeekToLast() {
S
Stanislau Hlebik 已提交
1296
  // Don't use iter_::Seek() if we set a prefix extractor
1297
  // because prefix seek will be used.
1298
  if (prefix_extractor_ != nullptr) {
S
Stanislau Hlebik 已提交
1299 1300
    max_skip_ = std::numeric_limits<uint64_t>::max();
  }
J
jorlow@chromium.org 已提交
1301
  direction_ = kReverse;
1302
  ReleaseTempPinnedData();
1303
  ResetInternalKeysSkippedCounter();
J
jorlow@chromium.org 已提交
1304
  ClearSavedValue();
1305 1306 1307 1308

  {
    PERF_TIMER_GUARD(seek_internal_seek_time);
    iter_->SeekToLast();
1309
    range_del_agg_.InvalidateTombstoneMapPositions();
1310
  }
1311 1312 1313 1314
  // 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) {
1315
    SeekForPrev(*iterate_upper_bound_);
1316
    range_del_agg_.InvalidateTombstoneMapPositions();
1317 1318 1319
    if (!Valid()) {
      return;
    } else if (user_comparator_->Equal(*iterate_upper_bound_, key())) {
1320 1321
      ReleaseTempPinnedData();
      PrevInternal();
1322
    }
1323 1324
  } else {
    PrevInternal();
1325
  }
M
Manuel Ung 已提交
1326 1327 1328 1329 1330
  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());
1331
      PERF_COUNTER_ADD(iter_read_bytes, key().size() + value().size());
M
Manuel Ung 已提交
1332 1333
    }
  }
1334
  if (valid_ && prefix_extractor_ && prefix_same_as_start_) {
1335 1336 1337
    prefix_start_buf_.SetUserKey(
        prefix_extractor_->Transform(saved_key_.GetUserKey()));
    prefix_start_key_ = prefix_start_buf_.GetUserKey();
1338
  }
J
jorlow@chromium.org 已提交
1339 1340
}

1341 1342 1343 1344 1345
Iterator* NewDBIterator(Env* env, const ReadOptions& read_options,
                        const ImmutableCFOptions& cf_options,
                        const Comparator* user_key_comparator,
                        InternalIterator* internal_iter,
                        const SequenceNumber& sequence,
Y
Yi Wu 已提交
1346
                        uint64_t max_sequential_skip_in_iterations,
Y
Yi Wu 已提交
1347 1348 1349 1350 1351
                        ReadCallback* read_callback, bool allow_blob) {
  DBIter* db_iter =
      new DBIter(env, read_options, cf_options, user_key_comparator,
                 internal_iter, sequence, false,
                 max_sequential_skip_in_iterations, read_callback, allow_blob);
1352
  return db_iter;
1353 1354
}

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

A
Andrew Kryczka 已提交
1357 1358 1359 1360
RangeDelAggregator* ArenaWrappedDBIter::GetRangeDelAggregator() {
  return db_iter_->GetRangeDelAggregator();
}

S
sdong 已提交
1361
void ArenaWrappedDBIter::SetIterUnderDBIter(InternalIterator* iter) {
1362 1363 1364 1365 1366 1367 1368 1369 1370
  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 已提交
1371 1372 1373
inline void ArenaWrappedDBIter::SeekForPrev(const Slice& target) {
  db_iter_->SeekForPrev(target);
}
1374 1375 1376 1377 1378
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 已提交
1379
bool ArenaWrappedDBIter::IsBlob() const { return db_iter_->IsBlob(); }
1380 1381
inline Status ArenaWrappedDBIter::GetProperty(std::string prop_name,
                                              std::string* prop) {
S
Siying Dong 已提交
1382 1383 1384 1385 1386 1387 1388
  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();
  }
1389 1390
  return db_iter_->GetProperty(prop_name, prop);
}
S
Siying Dong 已提交
1391 1392 1393 1394 1395

void ArenaWrappedDBIter::Init(Env* env, const ReadOptions& read_options,
                              const ImmutableCFOptions& cf_options,
                              const SequenceNumber& sequence,
                              uint64_t max_sequential_skip_in_iteration,
Y
Yi Wu 已提交
1396
                              uint64_t version_number,
1397 1398
                              ReadCallback* read_callback, bool allow_blob,
                              bool allow_refresh) {
S
Siying Dong 已提交
1399 1400 1401
  auto mem = arena_.AllocateAligned(sizeof(DBIter));
  db_iter_ = new (mem)
      DBIter(env, read_options, cf_options, cf_options.user_comparator, nullptr,
Y
Yi Wu 已提交
1402 1403
             sequence, true, max_sequential_skip_in_iteration, read_callback,
             allow_blob);
S
Siying Dong 已提交
1404
  sv_number_ = version_number;
1405
  allow_refresh_ = allow_refresh;
S
Siying Dong 已提交
1406 1407 1408
}

Status ArenaWrappedDBIter::Refresh() {
1409
  if (cfd_ == nullptr || db_impl_ == nullptr || !allow_refresh_) {
S
Siying Dong 已提交
1410 1411 1412
    return Status::NotSupported("Creating renew iterator is not allowed.");
  }
  assert(db_iter_ != nullptr);
1413 1414 1415
  // 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 已提交
1416 1417 1418 1419 1420 1421 1422 1423 1424 1425 1426
  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,
1427
         cur_sv_number, read_callback_, allow_blob_, allow_refresh_);
S
Siying Dong 已提交
1428 1429 1430 1431 1432 1433 1434 1435 1436

    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();
1437
}
J
jorlow@chromium.org 已提交
1438

1439
ArenaWrappedDBIter* NewArenaWrappedDbIterator(
1440
    Env* env, const ReadOptions& read_options,
S
Siying Dong 已提交
1441 1442
    const ImmutableCFOptions& cf_options, const SequenceNumber& sequence,
    uint64_t max_sequential_skip_in_iterations, uint64_t version_number,
Y
Yi Wu 已提交
1443
    ReadCallback* read_callback, DBImpl* db_impl, ColumnFamilyData* cfd,
1444
    bool allow_blob, bool allow_refresh) {
1445
  ArenaWrappedDBIter* iter = new ArenaWrappedDBIter();
S
Siying Dong 已提交
1446
  iter->Init(env, read_options, cf_options, sequence,
Y
Yi Wu 已提交
1447
             max_sequential_skip_in_iterations, version_number, read_callback,
1448 1449
             allow_blob, allow_refresh);
  if (db_impl != nullptr && cfd != nullptr && allow_refresh) {
Y
Yi Wu 已提交
1450 1451
    iter->StoreRefreshInfo(read_options, db_impl, cfd, read_callback,
                           allow_blob);
S
Siying Dong 已提交
1452
  }
1453

1454
  return iter;
J
jorlow@chromium.org 已提交
1455 1456
}

1457
}  // namespace rocksdb