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

34
namespace ROCKSDB_NAMESPACE {
J
jorlow@chromium.org 已提交
35

36
DBIter::DBIter(Env* _env, const ReadOptions& read_options,
37 38 39 40 41 42 43 44
               const ImmutableCFOptions& cf_options,
               const MutableCFOptions& mutable_cf_options,
               const Comparator* cmp, InternalIterator* iter, SequenceNumber s,
               bool arena_mode, uint64_t max_sequential_skip_in_iterations,
               ReadCallback* read_callback, DBImpl* db_impl,
               ColumnFamilyData* cfd, bool allow_blob)
    : prefix_extractor_(mutable_cf_options.prefix_extractor.get()),
      env_(_env),
45 46 47 48 49 50 51
      logger_(cf_options.info_log),
      user_comparator_(cmp),
      merge_operator_(cf_options.merge_operator),
      iter_(iter),
      read_callback_(read_callback),
      sequence_(s),
      statistics_(cf_options.statistics),
Y
Yanqin Jin 已提交
52 53
      max_skip_(max_sequential_skip_in_iterations),
      max_skippable_internal_keys_(read_options.max_skippable_internal_keys),
54 55 56 57 58 59 60
      num_internal_keys_skipped_(0),
      iterate_lower_bound_(read_options.iterate_lower_bound),
      iterate_upper_bound_(read_options.iterate_upper_bound),
      direction_(kForward),
      valid_(false),
      current_entry_is_merged_(false),
      is_key_seqnum_zero_(false),
61 62 63
      prefix_same_as_start_(mutable_cf_options.prefix_extractor
                                ? read_options.prefix_same_as_start
                                : false),
64
      pin_thru_lifetime_(read_options.pin_data),
S
sdong 已提交
65 66 67
      expect_total_order_inner_iter_(prefix_extractor_ == nullptr ||
                                     read_options.total_order_seek ||
                                     read_options.auto_prefix_mode),
68 69 70 71 72 73
      allow_blob_(allow_blob),
      is_blob_(false),
      arena_mode_(arena_mode),
      range_del_agg_(&cf_options.internal_comparator, s),
      db_impl_(db_impl),
      cfd_(cfd),
Y
Yanqin Jin 已提交
74 75
      start_seqnum_(read_options.iter_start_seqnum),
      timestamp_ub_(read_options.timestamp),
76
      timestamp_lb_(read_options.iter_start_ts),
Y
Yanqin Jin 已提交
77
      timestamp_size_(timestamp_ub_ ? timestamp_ub_->size() : 0) {
78 79 80 81 82
  RecordTick(statistics_, NO_ITERATOR_CREATED);
  if (pin_thru_lifetime_) {
    pinned_iters_mgr_.StartPinning();
  }
  if (iter_.iter()) {
83
    iter_.iter()->SetPinnedItersMgr(&pinned_iters_mgr_);
J
jorlow@chromium.org 已提交
84
  }
Y
Yanqin Jin 已提交
85
  assert(timestamp_size_ == user_comparator_.timestamp_size());
86
}
87

88 89 90
Status DBIter::GetProperty(std::string prop_name, std::string* prop) {
  if (prop == nullptr) {
    return Status::InvalidArgument("prop is nullptr");
91
  }
92 93 94 95
  if (prop_name == "rocksdb.iterator.super-version-number") {
    // First try to pass the value returned from inner iterator.
    return iter_.iter()->GetProperty(prop_name, prop);
  } else if (prop_name == "rocksdb.iterator.is-key-pinned") {
96
    if (valid_) {
97 98 99
      *prop = (pin_thru_lifetime_ && saved_key_.IsKeyPinned()) ? "1" : "0";
    } else {
      *prop = "Iterator is not valid.";
100
    }
101 102 103 104
    return Status::OK();
  } else if (prop_name == "rocksdb.iterator.internal-key") {
    *prop = saved_key_.GetUserKey().ToString();
    return Status::OK();
105
  }
106 107
  return Status::InvalidArgument("Unidentified property.");
}
108

109
bool DBIter::ParseKey(ParsedInternalKey* ikey) {
110
  if (!ParseInternalKey(iter_.key(), ikey)) {
J
jorlow@chromium.org 已提交
111
    status_ = Status::Corruption("corrupted internal key in DBIter");
112
    valid_ = false;
113
    ROCKS_LOG_ERROR(logger_, "corrupted internal key in DBIter: %s",
114
                    iter_.key().ToString(true).c_str());
J
jorlow@chromium.org 已提交
115 116 117 118 119 120
    return false;
  } else {
    return true;
  }
}

J
jorlow@chromium.org 已提交
121 122
void DBIter::Next() {
  assert(valid_);
123
  assert(status_.ok());
J
jorlow@chromium.org 已提交
124

125
  PERF_CPU_TIMER_GUARD(iter_next_cpu_nanos, env_);
126 127
  // Release temporarily pinned blocks from last operation
  ReleaseTempPinnedData();
128 129 130
  local_stats_.skip_count_ += num_internal_keys_skipped_;
  local_stats_.skip_count_--;
  num_internal_keys_skipped_ = 0;
131
  bool ok = true;
S
Stanislau Hlebik 已提交
132
  if (direction_ == kReverse) {
133
    is_key_seqnum_zero_ = false;
134 135 136
    if (!ReverseToForward()) {
      ok = false;
    }
137
  } else if (!current_entry_is_merged_) {
138 139 140 141 142
    // 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.
143 144
    assert(iter_.Valid());
    iter_.Next();
145
    PERF_COUNTER_ADD(internal_key_skipped_count, 1);
J
jorlow@chromium.org 已提交
146
  }
J
jorlow@chromium.org 已提交
147

148
  local_stats_.next_count_++;
149
  if (ok && iter_.Valid()) {
150 151 152 153 154
    Slice prefix;
    if (prefix_same_as_start_) {
      assert(prefix_extractor_ != nullptr);
      prefix = prefix_.GetUserKey();
    }
155
    FindNextUserEntry(true /* skipping the current user key */,
156
                      prefix_same_as_start_ ? &prefix : nullptr);
157
  } else {
158
    is_key_seqnum_zero_ = false;
159 160
    valid_ = false;
  }
161 162 163 164
  if (statistics_ != nullptr && valid_) {
    local_stats_.next_found_count_++;
    local_stats_.bytes_read_ += (key().size() + value().size());
  }
J
jorlow@chromium.org 已提交
165 166
}

167
// PRE: saved_key_ has the current user key if skipping_saved_key
168 169 170 171 172 173
// 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
174
//       a delete marker or a sequence number higher than sequence_
175
//       saved_key_ MUST have a proper user_key before calling this function
176
//
Y
Yanqin Jin 已提交
177
// The prefix parameter, if not null, indicates that we need to iterate
178 179 180
// within the prefix, and the iterator needs to be made invalid, if no
// more entry for the prefix can be found.
bool DBIter::FindNextUserEntry(bool skipping_saved_key, const Slice* prefix) {
181
  PERF_TIMER_GUARD(find_next_user_entry_time);
182
  return FindNextUserEntryInternal(skipping_saved_key, prefix);
183 184 185
}

// Actual implementation of DBIter::FindNextUserEntry()
186 187
bool DBIter::FindNextUserEntryInternal(bool skipping_saved_key,
                                       const Slice* prefix) {
J
jorlow@chromium.org 已提交
188
  // Loop until we hit an acceptable entry to yield
189
  assert(iter_.Valid());
190
  assert(status_.ok());
J
jorlow@chromium.org 已提交
191
  assert(direction_ == kForward);
192
  current_entry_is_merged_ = false;
193 194 195

  // 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
196
  // sequence numbers were too high or because skipping_saved_key = true.
197
  // What saved_key_ contains throughout this method:
Y
Yanqin Jin 已提交
198 199 200 201 202 203 204 205
  //  - if skipping_saved_key : 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.
206
  uint64_t num_skipped = 0;
207 208 209 210 211
  // For write unprepared, the target sequence number in reseek could be larger
  // than the snapshot, and thus needs to be skipped again. This could result in
  // an infinite loop of reseeks. To avoid that, we limit the number of reseeks
  // to one.
  bool reseek_done = false;
212

Y
Yi Wu 已提交
213 214
  is_blob_ = false;

J
jorlow@chromium.org 已提交
215
  do {
216 217 218
    // 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_;
219
    if (!ParseKey(&ikey_)) {
220
      is_key_seqnum_zero_ = false;
221
      return false;
222
    }
223

224 225
    is_key_seqnum_zero_ = (ikey_.sequence == 0);

226
    assert(iterate_upper_bound_ == nullptr || iter_.MayBeOutOfUpperBound() ||
Y
Yanqin Jin 已提交
227 228 229
           user_comparator_.CompareWithoutTimestamp(
               ikey_.user_key, /*a_has_ts=*/true, *iterate_upper_bound_,
               /*b_has_ts=*/false) < 0);
230
    if (iterate_upper_bound_ != nullptr && iter_.MayBeOutOfUpperBound() &&
Y
Yanqin Jin 已提交
231 232 233
        user_comparator_.CompareWithoutTimestamp(
            ikey_.user_key, /*a_has_ts=*/true, *iterate_upper_bound_,
            /*b_has_ts=*/false) >= 0) {
234 235
      break;
    }
236

237 238 239 240
    assert(prefix == nullptr || prefix_extractor_ != nullptr);
    if (prefix != nullptr &&
        prefix_extractor_->Transform(ikey_.user_key).compare(*prefix) != 0) {
      assert(prefix_same_as_start_);
241 242 243
      break;
    }

244
    if (TooManyInternalKeysSkipped()) {
245
      return false;
246 247
    }

Y
Yanqin Jin 已提交
248 249
    assert(ikey_.user_key.size() >= timestamp_size_);
    Slice ts;
250
    bool more_recent = false;
Y
Yanqin Jin 已提交
251 252 253
    if (timestamp_size_ > 0) {
      ts = ExtractTimestampFromUserKey(ikey_.user_key, timestamp_size_);
    }
254
    if (IsVisible(ikey_.sequence, ts, &more_recent)) {
255 256 257 258
      // 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.
259 260 261 262 263
      // Note that with current timestamp implementation, the same user key can
      // have different timestamps and zero sequence number on the bottommost
      // level. This will change in the future.
      if ((!is_prev_key_seqnum_zero || timestamp_size_ > 0) &&
          skipping_saved_key &&
264
          CompareKeyForSkip(ikey_.user_key, saved_key_.GetUserKey()) <= 0) {
265 266 267
        num_skipped++;  // skip this entry
        PERF_COUNTER_ADD(internal_key_skipped_count, 1);
      } else {
268
        assert(!skipping_saved_key ||
269
               CompareKeyForSkip(ikey_.user_key, saved_key_.GetUserKey()) > 0);
270 271 272 273 274
        if (!iter_.PrepareValue()) {
          assert(!iter_.status().ok());
          valid_ = false;
          return false;
        }
275
        num_skipped = 0;
276
        reseek_done = false;
277
        switch (ikey_.type) {
278 279 280 281
          case kTypeDeletion:
          case kTypeSingleDeletion:
            // Arrange to skip all upcoming entries for this key since
            // they are hidden by this deletion.
282 283 284
            // if iterartor specified start_seqnum we
            // 1) return internal key, including the type
            // 2) return ikey only if ikey.seqnum >= start_seqnum_
285
            // note that if deletion seqnum is < start_seqnum_ we
286 287 288
            // just skip it like in normal iterator.
            if (start_seqnum_ > 0 && ikey_.sequence >= start_seqnum_)  {
              saved_key_.SetInternalKey(ikey_);
289 290
              valid_ = true;
              return true;
291 292
            } else {
              saved_key_.SetUserKey(
293 294
                  ikey_.user_key, !pin_thru_lifetime_ ||
                                      !iter_.iter()->IsKeyPinned() /* copy */);
295
              skipping_saved_key = true;
296
              PERF_COUNTER_ADD(internal_delete_skipped_count, 1);
297
            }
298 299
            break;
          case kTypeValue:
Y
Yi Wu 已提交
300
          case kTypeBlobIndex:
301 302 303
            if (start_seqnum_ > 0) {
              // we are taking incremental snapshot here
              // incremental snapshots aren't supported on DB with range deletes
Y
Yanqin Jin 已提交
304
              assert(ikey_.type != kTypeBlobIndex);
305 306
              if (ikey_.sequence >= start_seqnum_) {
                saved_key_.SetInternalKey(ikey_);
Y
Yi Wu 已提交
307
                valid_ = true;
308
                return true;
309 310
              } else {
                // this key and all previous versions shouldn't be included,
311
                // skipping_saved_key
312 313 314 315
                saved_key_.SetUserKey(
                    ikey_.user_key,
                    !pin_thru_lifetime_ ||
                        !iter_.iter()->IsKeyPinned() /* copy */);
316
                skipping_saved_key = true;
Y
Yi Wu 已提交
317
              }
318
            } else {
319
              saved_key_.SetUserKey(
320 321
                  ikey_.user_key, !pin_thru_lifetime_ ||
                                      !iter_.iter()->IsKeyPinned() /* copy */);
322
              if (range_del_agg_.ShouldDelete(
323
                      ikey_, RangeDelPositioningMode::kForwardTraversal)) {
324 325
                // Arrange to skip all upcoming entries for this key since
                // they are hidden by this deletion.
326
                skipping_saved_key = true;
327
                num_skipped = 0;
328
                reseek_done = false;
329 330 331 332 333 334
                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 "
335
                      "ROCKSDB_NAMESPACE::blob_db::BlobDB instead.");
336
                  valid_ = false;
337
                  return false;
338
                }
339 340 341 342

                is_blob_ = true;
                valid_ = true;
                return true;
343 344
              } else {
                valid_ = true;
345
                return true;
346
              }
347 348 349
            }
            break;
          case kTypeMerge:
350
            saved_key_.SetUserKey(
351
                ikey_.user_key,
352
                !pin_thru_lifetime_ || !iter_.iter()->IsKeyPinned() /* copy */);
353
            if (range_del_agg_.ShouldDelete(
354
                    ikey_, RangeDelPositioningMode::kForwardTraversal)) {
355 356
              // Arrange to skip all upcoming entries for this key since
              // they are hidden by this deletion.
357
              skipping_saved_key = true;
358
              num_skipped = 0;
359
              reseek_done = false;
360 361 362 363 364 365
              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;
366
              return MergeValuesNewToOld();  // Go to a different state machine
367 368 369 370 371
            }
            break;
          default:
            assert(false);
            break;
372
        }
J
jorlow@chromium.org 已提交
373
      }
374
    } else {
375 376 377
      if (more_recent) {
        PERF_COUNTER_ADD(internal_recent_skipped_count, 1);
      }
378

379 380 381
      // This key was inserted after our snapshot was taken or skipped by
      // timestamp range. If this happens too many times in a row for the same
      // user key, we want to seek to the target sequence number.
Y
Yanqin Jin 已提交
382 383
      int cmp = user_comparator_.CompareWithoutTimestamp(
          ikey_.user_key, saved_key_.GetUserKey());
Y
Yanqin Jin 已提交
384
      if (cmp == 0 || (skipping_saved_key && cmp < 0)) {
385 386
        num_skipped++;
      } else {
387
        saved_key_.SetUserKey(
388
            ikey_.user_key,
389
            !iter_.iter()->IsKeyPinned() || !pin_thru_lifetime_ /* copy */);
390
        skipping_saved_key = false;
391
        num_skipped = 0;
392
        reseek_done = false;
393
      }
J
jorlow@chromium.org 已提交
394
    }
395 396 397

    // If we have sequentially iterated via numerous equal keys, then it's
    // better to seek so that we can avoid too many key comparisons.
398 399 400 401 402
    //
    // To avoid infinite loops, do not reseek if we have already attempted to
    // reseek previously.
    //
    // TODO(lth): If we reseek to sequence number greater than ikey_.sequence,
Y
Yanqin Jin 已提交
403
    // then it does not make sense to reseek as we would actually land further
404 405
    // away from the desired key. There is opportunity for optimization here.
    if (num_skipped > max_skip_ && !reseek_done) {
406
      is_key_seqnum_zero_ = false;
407
      num_skipped = 0;
408
      reseek_done = true;
409
      std::string last_key;
410
      if (skipping_saved_key) {
411 412 413
        // 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).
Y
Yanqin Jin 已提交
414 415 416 417 418 419 420 421 422 423 424
        if (timestamp_size_ == 0) {
          AppendInternalKey(
              &last_key,
              ParsedInternalKey(saved_key_.GetUserKey(), 0, kTypeDeletion));
        } else {
          std::string min_ts(timestamp_size_, static_cast<char>(0));
          AppendInternalKeyWithDifferentTimestamp(
              &last_key,
              ParsedInternalKey(saved_key_.GetUserKey(), 0, kTypeDeletion),
              min_ts);
        }
425 426
        // Don't set skipping_saved_key = false because we may still see more
        // user-keys equal to saved_key_.
427 428 429 430 431 432
      } 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.
Y
Yanqin Jin 已提交
433 434 435 436 437 438 439 440 441 442 443
        if (timestamp_size_ == 0) {
          AppendInternalKey(
              &last_key, ParsedInternalKey(saved_key_.GetUserKey(), sequence_,
                                           kValueTypeForSeek));
        } else {
          AppendInternalKeyWithDifferentTimestamp(
              &last_key,
              ParsedInternalKey(saved_key_.GetUserKey(), sequence_,
                                kValueTypeForSeek),
              *timestamp_ub_);
        }
444
      }
445
      iter_.Seek(last_key);
446 447
      RecordTick(statistics_, NUMBER_OF_RESEEKS_IN_ITERATION);
    } else {
448
      iter_.Next();
449
    }
450
  } while (iter_.Valid());
451

J
jorlow@chromium.org 已提交
452
  valid_ = false;
453
  return iter_.status().ok();
J
jorlow@chromium.org 已提交
454 455
}

456 457
// Merge values of the same user key starting from the current iter_ position
// Scan from the newer entries to older entries.
458
// PRE: iter_.key() points to the first merge type entry
459
//      saved_key_ stores the user key
460
//      iter_.PrepareValue() has been called
461 462
// POST: saved_value_ has the merged value for the user key
//       iter_ points to the next entry (or invalid)
463
bool DBIter::MergeValuesNewToOld() {
464
  if (!merge_operator_) {
465
    ROCKS_LOG_ERROR(logger_, "Options::merge_operator is null.");
466
    status_ = Status::InvalidArgument("merge_operator_ must be set.");
467
    valid_ = false;
468
    return false;
D
Deon Nicholas 已提交
469
  }
470

471 472
  // Temporarily pin the blocks that hold merge operands
  TempPinData();
473
  merge_context_.Clear();
474
  // Start the merge process by pushing the first operand
475 476
  merge_context_.PushOperand(
      iter_.value(), iter_.iter()->IsValuePinned() /* operand_pinned */);
477
  TEST_SYNC_POINT("DBIter::MergeValuesNewToOld:PushedFirstOperand");
478 479

  ParsedInternalKey ikey;
480
  Status s;
481
  for (iter_.Next(); iter_.Valid(); iter_.Next()) {
482
    TEST_SYNC_POINT("DBIter::MergeValuesNewToOld:SteppedToNextOperand");
483
    if (!ParseKey(&ikey)) {
484
      return false;
485 486
    }

487
    if (!user_comparator_.Equal(ikey.user_key, saved_key_.GetUserKey())) {
488 489
      // hit the next user key, stop right here
      break;
490 491
    }
    if (kTypeDeletion == ikey.type || kTypeSingleDeletion == ikey.type ||
492
               range_del_agg_.ShouldDelete(
493
                   ikey, RangeDelPositioningMode::kForwardTraversal)) {
494 495
      // hit a delete with the same user key, stop right here
      // iter_ is positioned after delete
496
      iter_.Next();
497
      break;
498 499 500 501 502 503 504
    }
    if (!iter_.PrepareValue()) {
      valid_ = false;
      return false;
    }

    if (kTypeValue == ikey.type) {
505 506
      // hit a put, merge the put value with operands and store the
      // final result in saved_value_. We are done!
507
      const Slice val = iter_.value();
508 509
      s = MergeHelper::TimedFullMerge(
          merge_operator_, ikey.user_key, &val, merge_context_.GetOperands(),
510
          &saved_value_, logger_, statistics_, env_, &pinned_value_, true);
511
      if (!s.ok()) {
Y
Yi Wu 已提交
512
        valid_ = false;
513
        status_ = s;
514
        return false;
515
      }
516
      // iter_ is positioned after put
517 518
      iter_.Next();
      if (!iter_.status().ok()) {
519 520 521 522
        valid_ = false;
        return false;
      }
      return true;
A
Andres Noetzli 已提交
523
    } else if (kTypeMerge == ikey.type) {
524 525
      // hit a merge, add the value as an operand and run associative merge.
      // when complete, add result to operands and continue.
526 527
      merge_context_.PushOperand(
          iter_.value(), iter_.iter()->IsValuePinned() /* operand_pinned */);
528
      PERF_COUNTER_ADD(internal_merge_count, 1);
Y
Yi Wu 已提交
529 530 531 532 533
    } 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 "
534
            "ROCKSDB_NAMESPACE::blob_db::BlobDB instead.");
Y
Yi Wu 已提交
535 536 537 538 539
      } else {
        status_ =
            Status::NotSupported("Blob DB does not support merge operator.");
      }
      valid_ = false;
540
      return false;
A
Andres Noetzli 已提交
541 542
    } else {
      assert(false);
543 544 545
    }
  }

546
  if (!iter_.status().ok()) {
547 548 549 550
    valid_ = false;
    return false;
  }

551 552 553 554
  // 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.
555 556 557
  s = MergeHelper::TimedFullMerge(merge_operator_, saved_key_.GetUserKey(),
                                  nullptr, merge_context_.GetOperands(),
                                  &saved_value_, logger_, statistics_, env_,
558
                                  &pinned_value_, true);
559
  if (!s.ok()) {
Y
Yi Wu 已提交
560
    valid_ = false;
561
    status_ = s;
562
    return false;
563
  }
564 565 566

  assert(status_.ok());
  return true;
567 568
}

J
jorlow@chromium.org 已提交
569
void DBIter::Prev() {
Y
Yanqin Jin 已提交
570 571 572 573 574 575 576
  if (timestamp_size_ > 0) {
    valid_ = false;
    status_ = Status::NotSupported(
        "SeekToLast/SeekForPrev/Prev currently not supported with timestamp.");
    return;
  }

J
jorlow@chromium.org 已提交
577
  assert(valid_);
578
  assert(status_.ok());
579 580

  PERF_CPU_TIMER_GUARD(iter_prev_cpu_nanos, env_);
581
  ReleaseTempPinnedData();
582
  ResetInternalKeysSkippedCounter();
583
  bool ok = true;
S
Stanislau Hlebik 已提交
584
  if (direction_ == kForward) {
585 586 587 588 589
    if (!ReverseToBackward()) {
      ok = false;
    }
  }
  if (ok) {
590 591 592 593 594 595
    Slice prefix;
    if (prefix_same_as_start_) {
      assert(prefix_extractor_ != nullptr);
      prefix = prefix_.GetUserKey();
    }
    PrevInternal(prefix_same_as_start_ ? &prefix : nullptr);
S
Stanislau Hlebik 已提交
596
  }
597

M
Manuel Ung 已提交
598
  if (statistics_ != nullptr) {
599
    local_stats_.prev_count_++;
M
Manuel Ung 已提交
600
    if (valid_) {
601 602
      local_stats_.prev_found_count_++;
      local_stats_.bytes_read_ += (key().size() + value().size());
M
Manuel Ung 已提交
603 604
    }
  }
S
Stanislau Hlebik 已提交
605
}
J
jorlow@chromium.org 已提交
606

607
bool DBIter::ReverseToForward() {
608
  assert(iter_.status().ok());
609 610 611 612

  // 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.
S
sdong 已提交
613
  if (!expect_total_order_inner_iter() || !iter_.Valid()) {
614 615
    IterKey last_key;
    last_key.SetInternalKey(ParsedInternalKey(
616
        saved_key_.GetUserKey(), kMaxSequenceNumber, kValueTypeForSeek));
617
    iter_.Seek(last_key.GetInternalKey());
618
  }
619

620
  direction_ = kForward;
621
  // Skip keys less than the current key() (a.k.a. saved_key_).
622
  while (iter_.Valid()) {
623 624 625 626
    ParsedInternalKey ikey;
    if (!ParseKey(&ikey)) {
      return false;
    }
627
    if (user_comparator_.Compare(ikey.user_key, saved_key_.GetUserKey()) >= 0) {
628 629
      return true;
    }
630
    iter_.Next();
631 632
  }

633
  if (!iter_.status().ok()) {
634 635
    valid_ = false;
    return false;
636
  }
637 638

  return true;
639 640
}

641 642
// Move iter_ to the key before saved_key_.
bool DBIter::ReverseToBackward() {
643
  assert(iter_.status().ok());
644 645 646 647 648

  // 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_ &&
S
sdong 已提交
649
      (!expect_total_order_inner_iter() || !iter_.Valid())) {
650
    IterKey last_key;
651 652 653 654 655
    // 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));
S
sdong 已提交
656
    if (!expect_total_order_inner_iter()) {
657
      iter_.SeekForPrev(last_key.GetInternalKey());
658 659 660 661 662
    } 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.
663 664 665
      iter_.Seek(last_key.GetInternalKey());
      if (!iter_.Valid() && iter_.status().ok()) {
        iter_.SeekToLast();
666
      }
667 668 669 670
    }
  }

  direction_ = kReverse;
671
  return FindUserKeyBeforeSavedKey();
672 673
}

674
void DBIter::PrevInternal(const Slice* prefix) {
675
  while (iter_.Valid()) {
676
    saved_key_.SetUserKey(
677 678
        ExtractUserKey(iter_.key()),
        !iter_.iter()->IsKeyPinned() || !pin_thru_lifetime_ /* copy */);
679

680 681
    assert(prefix == nullptr || prefix_extractor_ != nullptr);
    if (prefix != nullptr &&
682
        prefix_extractor_->Transform(saved_key_.GetUserKey())
683 684
                .compare(*prefix) != 0) {
      assert(prefix_same_as_start_);
685 686 687 688 689
      // Current key does not have the same prefix as start
      valid_ = false;
      return;
    }

690 691 692 693
    assert(iterate_lower_bound_ == nullptr || iter_.MayBeOutOfLowerBound() ||
           user_comparator_.Compare(saved_key_.GetUserKey(),
                                    *iterate_lower_bound_) >= 0);
    if (iterate_lower_bound_ != nullptr && iter_.MayBeOutOfLowerBound() &&
694 695
        user_comparator_.Compare(saved_key_.GetUserKey(),
                                 *iterate_lower_bound_) < 0) {
696 697 698 699 700
      // We've iterated earlier than the user-specified lower bound.
      valid_ = false;
      return;
    }

701
    if (!FindValueForCurrentKey()) {  // assigns valid_
S
Stanislau Hlebik 已提交
702
      return;
J
jorlow@chromium.org 已提交
703
    }
704

705 706 707
    // Whether or not we found a value for current key, we need iter_ to end up
    // on a smaller key.
    if (!FindUserKeyBeforeSavedKey()) {
708 709 710
      return;
    }

711 712 713
    if (valid_) {
      // Found the value.
      return;
S
Stanislau Hlebik 已提交
714
    }
715 716 717

    if (TooManyInternalKeysSkipped(false)) {
      return;
S
Stanislau Hlebik 已提交
718 719
    }
  }
720

S
Stanislau Hlebik 已提交
721 722
  // We haven't found any key - iterator is not valid
  valid_ = false;
J
jorlow@chromium.org 已提交
723 724
}

725 726 727 728 729 730 731 732 733 734 735
// 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 已提交
736
bool DBIter::FindValueForCurrentKey() {
737
  assert(iter_.Valid());
738
  merge_context_.Clear();
739
  current_entry_is_merged_ = false;
A
Andres Noetzli 已提交
740 741
  // last entry before merge (could be kTypeDeletion, kTypeSingleDeletion or
  // kTypeValue)
S
Stanislau Hlebik 已提交
742 743
  ValueType last_not_merge_type = kTypeDeletion;
  ValueType last_key_entry_type = kTypeDeletion;
J
jorlow@chromium.org 已提交
744

745 746 747
  // Temporarily pin blocks that hold (merge operands / the value)
  ReleaseTempPinnedData();
  TempPinData();
S
Stanislau Hlebik 已提交
748
  size_t num_skipped = 0;
749
  while (iter_.Valid()) {
750 751 752 753 754
    ParsedInternalKey ikey;
    if (!ParseKey(&ikey)) {
      return false;
    }

Y
Yanqin Jin 已提交
755 756 757 758 759 760 761
    assert(ikey.user_key.size() >= timestamp_size_);
    Slice ts;
    if (timestamp_size_ > 0) {
      ts = Slice(ikey.user_key.data() + ikey.user_key.size() - timestamp_size_,
                 timestamp_size_);
    }
    if (!IsVisible(ikey.sequence, ts) ||
762
        !user_comparator_.Equal(ikey.user_key, saved_key_.GetUserKey())) {
763 764
      break;
    }
765 766 767 768
    if (TooManyInternalKeysSkipped()) {
      return false;
    }

769 770 771
    // 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.
772
    if (num_skipped >= max_skip_) {
S
Stanislau Hlebik 已提交
773 774 775
      return FindValueForCurrentKeyUsingSeek();
    }

776 777 778 779 780
    if (!iter_.PrepareValue()) {
      valid_ = false;
      return false;
    }

S
Stanislau Hlebik 已提交
781 782 783
    last_key_entry_type = ikey.type;
    switch (last_key_entry_type) {
      case kTypeValue:
Y
Yi Wu 已提交
784
      case kTypeBlobIndex:
785
        if (range_del_agg_.ShouldDelete(
786
                ikey, RangeDelPositioningMode::kBackwardTraversal)) {
A
Andrew Kryczka 已提交
787 788 789
          last_key_entry_type = kTypeRangeDeletion;
          PERF_COUNTER_ADD(internal_delete_skipped_count, 1);
        } else {
790 791
          assert(iter_.iter()->IsValuePinned());
          pinned_value_ = iter_.value();
A
Andrew Kryczka 已提交
792
        }
793
        merge_context_.Clear();
A
Andrew Kryczka 已提交
794
        last_not_merge_type = last_key_entry_type;
S
Stanislau Hlebik 已提交
795 796
        break;
      case kTypeDeletion:
A
Andres Noetzli 已提交
797
      case kTypeSingleDeletion:
798
        merge_context_.Clear();
A
Andres Noetzli 已提交
799
        last_not_merge_type = last_key_entry_type;
800
        PERF_COUNTER_ADD(internal_delete_skipped_count, 1);
S
Stanislau Hlebik 已提交
801 802
        break;
      case kTypeMerge:
803
        if (range_del_agg_.ShouldDelete(
804
                ikey, RangeDelPositioningMode::kBackwardTraversal)) {
A
Andrew Kryczka 已提交
805 806 807 808 809 810 811
          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(
812 813
              iter_.value(),
              iter_.iter()->IsValuePinned() /* operand_pinned */);
814
          PERF_COUNTER_ADD(internal_merge_count, 1);
A
Andrew Kryczka 已提交
815
        }
S
Stanislau Hlebik 已提交
816 817 818 819 820
        break;
      default:
        assert(false);
    }

821
    PERF_COUNTER_ADD(internal_key_skipped_count, 1);
822
    iter_.Prev();
S
Stanislau Hlebik 已提交
823
    ++num_skipped;
824 825
  }

826
  if (!iter_.status().ok()) {
827 828
    valid_ = false;
    return false;
S
Stanislau Hlebik 已提交
829 830
  }

831
  Status s;
Y
Yi Wu 已提交
832
  is_blob_ = false;
S
Stanislau Hlebik 已提交
833 834
  switch (last_key_entry_type) {
    case kTypeDeletion:
A
Andres Noetzli 已提交
835
    case kTypeSingleDeletion:
A
Andrew Kryczka 已提交
836
    case kTypeRangeDeletion:
S
Stanislau Hlebik 已提交
837
      valid_ = false;
838
      return true;
S
Stanislau Hlebik 已提交
839
    case kTypeMerge:
840
      current_entry_is_merged_ = true;
A
Aaron Gao 已提交
841
      if (last_not_merge_type == kTypeDeletion ||
A
Andrew Kryczka 已提交
842 843
          last_not_merge_type == kTypeSingleDeletion ||
          last_not_merge_type == kTypeRangeDeletion) {
844 845 846
        s = MergeHelper::TimedFullMerge(
            merge_operator_, saved_key_.GetUserKey(), nullptr,
            merge_context_.GetOperands(), &saved_value_, logger_, statistics_,
847
            env_, &pinned_value_, true);
Y
Yi Wu 已提交
848 849 850 851 852
      } 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 "
853
              "ROCKSDB_NAMESPACE::blob_db::BlobDB instead.");
Y
Yi Wu 已提交
854 855 856 857 858
        } else {
          status_ =
              Status::NotSupported("Blob DB does not support merge operator.");
        }
        valid_ = false;
859
        return false;
860
      } else {
S
Stanislau Hlebik 已提交
861
        assert(last_not_merge_type == kTypeValue);
862
        s = MergeHelper::TimedFullMerge(
863
            merge_operator_, saved_key_.GetUserKey(), &pinned_value_,
864
            merge_context_.GetOperands(), &saved_value_, logger_, statistics_,
865
            env_, &pinned_value_, true);
866
      }
S
Stanislau Hlebik 已提交
867 868
      break;
    case kTypeValue:
869
      // do nothing - we've already has value in pinned_value_
S
Stanislau Hlebik 已提交
870
      break;
Y
Yi Wu 已提交
871 872 873 874 875
    case kTypeBlobIndex:
      if (!allow_blob_) {
        ROCKS_LOG_ERROR(logger_, "Encounter unexpected blob index.");
        status_ = Status::NotSupported(
            "Encounter unexpected blob index. Please open DB with "
876
            "ROCKSDB_NAMESPACE::blob_db::BlobDB instead.");
Y
Yi Wu 已提交
877
        valid_ = false;
878
        return false;
Y
Yi Wu 已提交
879 880 881
      }
      is_blob_ = true;
      break;
S
Stanislau Hlebik 已提交
882 883 884
    default:
      assert(false);
      break;
J
jorlow@chromium.org 已提交
885
  }
886
  if (!s.ok()) {
Y
Yi Wu 已提交
887
    valid_ = false;
888
    status_ = s;
889
    return false;
890
  }
891
  valid_ = true;
S
Stanislau Hlebik 已提交
892 893
  return true;
}
J
jorlow@chromium.org 已提交
894

S
Stanislau Hlebik 已提交
895 896
// This function is used in FindValueForCurrentKey.
// We use Seek() function instead of Prev() to find necessary value
897 898
// TODO: This is very similar to FindNextUserEntry() and MergeValuesNewToOld().
//       Would be nice to reuse some code.
S
Stanislau Hlebik 已提交
899
bool DBIter::FindValueForCurrentKeyUsingSeek() {
900 901 902
  // FindValueForCurrentKey will enable pinning before calling
  // FindValueForCurrentKeyUsingSeek()
  assert(pinned_iters_mgr_.PinningEnabled());
S
Stanislau Hlebik 已提交
903
  std::string last_key;
904 905
  AppendInternalKey(&last_key, ParsedInternalKey(saved_key_.GetUserKey(),
                                                 sequence_, kValueTypeForSeek));
906
  iter_.Seek(last_key);
S
Stanislau Hlebik 已提交
907 908
  RecordTick(statistics_, NUMBER_OF_RESEEKS_IN_ITERATION);

909 910
  // In case read_callback presents, the value we seek to may not be visible.
  // Find the next value that's visible.
S
Stanislau Hlebik 已提交
911
  ParsedInternalKey ikey;
912
  is_blob_ = false;
913
  while (true) {
914
    if (!iter_.Valid()) {
915
      valid_ = false;
916
      return iter_.status().ok();
917 918 919 920 921
    }

    if (!ParseKey(&ikey)) {
      return false;
    }
Y
Yanqin Jin 已提交
922 923 924 925 926 927 928
    assert(ikey.user_key.size() >= timestamp_size_);
    Slice ts;
    if (timestamp_size_ > 0) {
      ts = Slice(ikey.user_key.data() + ikey.user_key.size() - timestamp_size_,
                 timestamp_size_);
    }

929
    if (!user_comparator_.Equal(ikey.user_key, saved_key_.GetUserKey())) {
930 931 932 933 934 935 936
      // 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;
    }

Y
Yanqin Jin 已提交
937
    if (IsVisible(ikey.sequence, ts)) {
938 939
      break;
    }
940

941
    iter_.Next();
942
  }
S
Stanislau Hlebik 已提交
943

A
Andrew Kryczka 已提交
944
  if (ikey.type == kTypeDeletion || ikey.type == kTypeSingleDeletion ||
945
      range_del_agg_.ShouldDelete(
946
          ikey, RangeDelPositioningMode::kBackwardTraversal)) {
J
jorlow@chromium.org 已提交
947
    valid_ = false;
948
    return true;
S
Stanislau Hlebik 已提交
949
  }
Y
Yi Wu 已提交
950 951 952 953
  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 "
954
        "ROCKSDB_NAMESPACE::blob_db::BlobDB instead.");
Y
Yi Wu 已提交
955
    valid_ = false;
956
    return false;
Y
Yi Wu 已提交
957
  }
958 959 960 961
  if (!iter_.PrepareValue()) {
    valid_ = false;
    return false;
  }
Y
Yi Wu 已提交
962
  if (ikey.type == kTypeValue || ikey.type == kTypeBlobIndex) {
963 964
    assert(iter_.iter()->IsValuePinned());
    pinned_value_ = iter_.value();
965
    is_blob_ = (ikey.type == kTypeBlobIndex);
A
Andrew Kryczka 已提交
966 967 968
    valid_ = true;
    return true;
  }
S
Stanislau Hlebik 已提交
969 970 971

  // kTypeMerge. We need to collect all kTypeMerge values and save them
  // in operands
972
  assert(ikey.type == kTypeMerge);
973
  current_entry_is_merged_ = true;
974
  merge_context_.Clear();
975 976
  merge_context_.PushOperand(
      iter_.value(), iter_.iter()->IsValuePinned() /* operand_pinned */);
977
  while (true) {
978
    iter_.Next();
S
Stanislau Hlebik 已提交
979

980 981
    if (!iter_.Valid()) {
      if (!iter_.status().ok()) {
982 983 984 985
        valid_ = false;
        return false;
      }
      break;
S
Stanislau Hlebik 已提交
986
    }
987 988 989
    if (!ParseKey(&ikey)) {
      return false;
    }
990
    if (!user_comparator_.Equal(ikey.user_key, saved_key_.GetUserKey())) {
991 992 993 994
      break;
    }
    if (ikey.type == kTypeDeletion || ikey.type == kTypeSingleDeletion ||
        range_del_agg_.ShouldDelete(
995
            ikey, RangeDelPositioningMode::kForwardTraversal)) {
996
      break;
997 998 999 1000 1001 1002 1003
    }
    if (!iter_.PrepareValue()) {
      valid_ = false;
      return false;
    }

    if (ikey.type == kTypeValue) {
1004
      const Slice val = iter_.value();
1005 1006 1007 1008 1009 1010 1011 1012 1013
      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 已提交
1014
      valid_ = true;
1015 1016
      return true;
    } else if (ikey.type == kTypeMerge) {
1017 1018
      merge_context_.PushOperand(
          iter_.value(), iter_.iter()->IsValuePinned() /* operand_pinned */);
1019 1020 1021 1022 1023 1024
      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 "
1025
            "ROCKSDB_NAMESPACE::blob_db::BlobDB instead.");
1026 1027 1028 1029
      } else {
        status_ =
            Status::NotSupported("Blob DB does not support merge operator.");
      }
Y
Yi Wu 已提交
1030
      valid_ = false;
1031 1032 1033
      return false;
    } else {
      assert(false);
1034
    }
S
Stanislau Hlebik 已提交
1035 1036
  }

1037 1038 1039 1040 1041
  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 已提交
1042
    valid_ = false;
1043
    status_ = s;
1044
    return false;
1045
  }
S
Stanislau Hlebik 已提交
1046

1047 1048 1049
  // 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.
S
sdong 已提交
1050 1051
  if (!expect_total_order_inner_iter() || !iter_.Valid()) {
    if (!expect_total_order_inner_iter()) {
1052
      iter_.SeekForPrev(last_key);
1053
    } else {
1054 1055 1056
      iter_.Seek(last_key);
      if (!iter_.Valid() && iter_.status().ok()) {
        iter_.SeekToLast();
1057 1058 1059
      }
    }
    RecordTick(statistics_, NUMBER_OF_RESEEKS_IN_ITERATION);
S
Stanislau Hlebik 已提交
1060
  }
1061 1062 1063

  valid_ = true;
  return true;
S
Stanislau Hlebik 已提交
1064 1065
}

1066 1067 1068 1069
// 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 已提交
1070
  size_t num_skipped = 0;
1071
  while (iter_.Valid()) {
1072 1073 1074
    ParsedInternalKey ikey;
    if (!ParseKey(&ikey)) {
      return false;
1075 1076
    }

1077
    if (user_comparator_.Compare(ikey.user_key, saved_key_.GetUserKey()) < 0) {
1078 1079 1080 1081 1082
      return true;
    }

    if (TooManyInternalKeysSkipped()) {
      return false;
S
Stanislau Hlebik 已提交
1083
    }
1084

S
Siying Dong 已提交
1085
    assert(ikey.sequence != kMaxSequenceNumber);
Y
Yanqin Jin 已提交
1086 1087 1088 1089 1090 1091 1092
    assert(ikey.user_key.size() >= timestamp_size_);
    Slice ts;
    if (timestamp_size_ > 0) {
      ts = Slice(ikey.user_key.data() + ikey.user_key.size() - timestamp_size_,
                 timestamp_size_);
    }
    if (!IsVisible(ikey.sequence, ts)) {
1093 1094 1095 1096
      PERF_COUNTER_ADD(internal_recent_skipped_count, 1);
    } else {
      PERF_COUNTER_ADD(internal_key_skipped_count, 1);
    }
1097

1098
    if (num_skipped >= max_skip_) {
1099 1100 1101 1102 1103 1104
      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.
1105
      iter_.Seek(last_key.GetInternalKey());
1106
      RecordTick(statistics_, NUMBER_OF_RESEEKS_IN_ITERATION);
1107
      if (!iter_.Valid()) {
1108 1109 1110 1111 1112 1113
        break;
      }
    } else {
      ++num_skipped;
    }

1114
    iter_.Prev();
S
Stanislau Hlebik 已提交
1115
  }
1116

1117
  if (!iter_.status().ok()) {
1118 1119 1120 1121 1122
    valid_ = false;
    return false;
  }

  return true;
S
Stanislau Hlebik 已提交
1123 1124
}

1125 1126 1127 1128 1129 1130 1131 1132 1133 1134 1135 1136
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;
}

1137 1138
bool DBIter::IsVisible(SequenceNumber sequence, const Slice& ts,
                       bool* more_recent) {
Y
Yanqin Jin 已提交
1139
  // Remember that comparator orders preceding timestamp as larger.
1140 1141 1142 1143 1144 1145 1146 1147 1148 1149 1150 1151 1152
  // TODO(yanqin): support timestamp in read_callback_.
  bool visible_by_seq = (read_callback_ == nullptr)
                            ? sequence <= sequence_
                            : read_callback_->IsVisible(sequence);

  bool visible_by_ts =
      (timestamp_ub_ == nullptr ||
       user_comparator_.CompareTimestamp(ts, *timestamp_ub_) <= 0) &&
      (timestamp_lb_ == nullptr ||
       user_comparator_.CompareTimestamp(ts, *timestamp_lb_) >= 0);

  if (more_recent) {
    *more_recent = !visible_by_seq;
1153
  }
1154
  return visible_by_seq && visible_by_ts;
1155
}
1156

1157
void DBIter::SetSavedKeyToSeekTarget(const Slice& target) {
1158
  is_key_seqnum_zero_ = false;
1159
  SequenceNumber seq = sequence_;
1160
  saved_key_.Clear();
Y
Yanqin Jin 已提交
1161
  saved_key_.SetInternalKey(target, seq, kValueTypeForSeek, timestamp_ub_);
1162

Z
zhangjinpeng1987 已提交
1163
  if (iterate_lower_bound_ != nullptr &&
Y
Yanqin Jin 已提交
1164 1165 1166
      user_comparator_.CompareWithoutTimestamp(
          saved_key_.GetUserKey(), /*a_has_ts=*/true, *iterate_lower_bound_,
          /*b_has_ts=*/false) < 0) {
1167
    // Seek key is smaller than the lower bound.
Z
zhangjinpeng1987 已提交
1168
    saved_key_.Clear();
Y
Yanqin Jin 已提交
1169 1170
    saved_key_.SetInternalKey(*iterate_lower_bound_, seq, kValueTypeForSeek,
                              timestamp_ub_);
Z
zhangjinpeng1987 已提交
1171
  }
1172 1173 1174 1175 1176 1177 1178 1179 1180 1181 1182 1183 1184 1185 1186 1187 1188 1189 1190 1191 1192 1193 1194 1195 1196 1197 1198 1199 1200 1201
}

void DBIter::SetSavedKeyToSeekForPrevTarget(const Slice& target) {
  is_key_seqnum_zero_ = false;
  saved_key_.Clear();
  // now saved_key is used to store internal key.
  saved_key_.SetInternalKey(target, 0 /* sequence_number */,
                            kValueTypeForSeekForPrev);

  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);
  }
}

void DBIter::Seek(const Slice& target) {
  PERF_CPU_TIMER_GUARD(iter_seek_cpu_nanos, env_);
  StopWatch sw(env_, statistics_, DB_SEEK);

#ifndef ROCKSDB_LITE
  if (db_impl_ != nullptr && cfd_ != nullptr) {
    db_impl_->TraceIteratorSeek(cfd_->GetID(), target);
  }
#endif  // ROCKSDB_LITE

  status_ = Status::OK();
  ReleaseTempPinnedData();
  ResetInternalKeysSkippedCounter();
Z
zhangjinpeng1987 已提交
1202

1203
  // Seek the inner iterator based on the target key.
1204 1205
  {
    PERF_TIMER_GUARD(seek_internal_seek_time);
1206 1207

    SetSavedKeyToSeekTarget(target);
1208
    iter_.Seek(saved_key_.GetInternalKey());
1209

1210
    range_del_agg_.InvalidateRangeDelMapPositions();
1211
    RecordTick(statistics_, NUMBER_DB_SEEK);
1212
  }
1213 1214 1215 1216 1217 1218 1219 1220 1221 1222 1223 1224 1225
  if (!iter_.Valid()) {
    valid_ = false;
    return;
  }
  direction_ = kForward;

  // Now the inner iterator is placed to the target position. From there,
  // we need to find out the next key that is visible to the user.
  ClearSavedValue();
  if (prefix_same_as_start_) {
    // The case where the iterator needs to be invalidated if it has exausted
    // keys within the same prefix of the seek key.
    assert(prefix_extractor_ != nullptr);
Y
Yanqin Jin 已提交
1226
    Slice target_prefix = prefix_extractor_->Transform(target);
1227 1228 1229
    FindNextUserEntry(false /* not skipping saved_key */,
                      &target_prefix /* prefix */);
    if (valid_) {
Y
Yanqin Jin 已提交
1230
      // Remember the prefix of the seek key for the future Next() call to
1231 1232
      // check.
      prefix_.SetUserKey(target_prefix);
M
Manuel Ung 已提交
1233
    }
J
jorlow@chromium.org 已提交
1234
  } else {
1235 1236 1237 1238
    FindNextUserEntry(false /* not skipping saved_key */, nullptr);
  }
  if (!valid_) {
    return;
J
jorlow@chromium.org 已提交
1239
  }
1240

1241 1242 1243 1244 1245
  // Updating stats and perf context counters.
  if (statistics_ != nullptr) {
    // Decrement since we don't want to count this key as skipped
    RecordTick(statistics_, NUMBER_DB_SEEK_FOUND);
    RecordTick(statistics_, ITER_BYTES_READ, key().size() + value().size());
1246
  }
1247
  PERF_COUNTER_ADD(iter_read_bytes, key().size() + value().size());
J
jorlow@chromium.org 已提交
1248
}
J
jorlow@chromium.org 已提交
1249

A
Aaron Gao 已提交
1250
void DBIter::SeekForPrev(const Slice& target) {
1251
  PERF_CPU_TIMER_GUARD(iter_seek_cpu_nanos, env_);
A
Aaron Gao 已提交
1252
  StopWatch sw(env_, statistics_, DB_SEEK);
1253 1254 1255 1256 1257 1258 1259

#ifndef ROCKSDB_LITE
  if (db_impl_ != nullptr && cfd_ != nullptr) {
    db_impl_->TraceIteratorSeekForPrev(cfd_->GetID(), target);
  }
#endif  // ROCKSDB_LITE

Y
Yanqin Jin 已提交
1260 1261 1262 1263 1264 1265 1266
  if (timestamp_size_ > 0) {
    valid_ = false;
    status_ = Status::NotSupported(
        "SeekToLast/SeekForPrev/Prev currently not supported with timestamp.");
    return;
  }

1267
  status_ = Status::OK();
A
Aaron Gao 已提交
1268
  ReleaseTempPinnedData();
1269
  ResetInternalKeysSkippedCounter();
Z
zhangjinpeng1987 已提交
1270

1271
  // Seek the inner iterator based on the target key.
A
Aaron Gao 已提交
1272 1273
  {
    PERF_TIMER_GUARD(seek_internal_seek_time);
1274
    SetSavedKeyToSeekForPrevTarget(target);
1275
    iter_.SeekForPrev(saved_key_.GetInternalKey());
1276
    range_del_agg_.InvalidateRangeDelMapPositions();
1277
    RecordTick(statistics_, NUMBER_DB_SEEK);
A
Aaron Gao 已提交
1278
  }
1279 1280 1281
  if (!iter_.Valid()) {
    valid_ = false;
    return;
1282
  }
1283
  direction_ = kReverse;
1284

1285 1286 1287 1288 1289 1290 1291 1292
  // Now the inner iterator is placed to the target position. From there,
  // we need to find out the first key that is visible to the user in the
  // backward direction.
  ClearSavedValue();
  if (prefix_same_as_start_) {
    // The case where the iterator needs to be invalidated if it has exausted
    // keys within the same prefix of the seek key.
    assert(prefix_extractor_ != nullptr);
Y
Yanqin Jin 已提交
1293
    Slice target_prefix = prefix_extractor_->Transform(target);
1294 1295 1296 1297 1298
    PrevInternal(&target_prefix);
    if (valid_) {
      // Remember the prefix of the seek key for the future Prev() call to
      // check.
      prefix_.SetUserKey(target_prefix);
A
Aaron Gao 已提交
1299 1300
    }
  } else {
1301
    PrevInternal(nullptr);
A
Aaron Gao 已提交
1302
  }
1303 1304 1305 1306 1307 1308

  // Report stats and perf context.
  if (statistics_ != nullptr && valid_) {
    RecordTick(statistics_, NUMBER_DB_SEEK_FOUND);
    RecordTick(statistics_, ITER_BYTES_READ, key().size() + value().size());
    PERF_COUNTER_ADD(iter_read_bytes, key().size() + value().size());
A
Aaron Gao 已提交
1309 1310 1311
  }
}

J
jorlow@chromium.org 已提交
1312
void DBIter::SeekToFirst() {
1313 1314 1315 1316
  if (iterate_lower_bound_ != nullptr) {
    Seek(*iterate_lower_bound_);
    return;
  }
1317
  PERF_CPU_TIMER_GUARD(iter_seek_cpu_nanos, env_);
1318 1319
  // Don't use iter_::Seek() if we set a prefix extractor
  // because prefix seek will be used.
S
sdong 已提交
1320
  if (!expect_total_order_inner_iter()) {
1321 1322 1323
    max_skip_ = std::numeric_limits<uint64_t>::max();
  }
  status_ = Status::OK();
J
jorlow@chromium.org 已提交
1324
  direction_ = kForward;
1325
  ReleaseTempPinnedData();
1326
  ResetInternalKeysSkippedCounter();
J
jorlow@chromium.org 已提交
1327
  ClearSavedValue();
1328
  is_key_seqnum_zero_ = false;
1329 1330 1331

  {
    PERF_TIMER_GUARD(seek_internal_seek_time);
1332
    iter_.SeekToFirst();
1333
    range_del_agg_.InvalidateRangeDelMapPositions();
1334 1335
  }

M
Manuel Ung 已提交
1336
  RecordTick(statistics_, NUMBER_DB_SEEK);
1337
  if (iter_.Valid()) {
1338
    saved_key_.SetUserKey(
1339 1340
        ExtractUserKey(iter_.key()),
        !iter_.iter()->IsKeyPinned() || !pin_thru_lifetime_ /* copy */);
1341 1342
    FindNextUserEntry(false /* not skipping saved_key */,
                      nullptr /* no prefix check */);
M
Manuel Ung 已提交
1343 1344 1345 1346
    if (statistics_ != nullptr) {
      if (valid_) {
        RecordTick(statistics_, NUMBER_DB_SEEK_FOUND);
        RecordTick(statistics_, ITER_BYTES_READ, key().size() + value().size());
1347
        PERF_COUNTER_ADD(iter_read_bytes, key().size() + value().size());
M
Manuel Ung 已提交
1348 1349
      }
    }
J
jorlow@chromium.org 已提交
1350 1351
  } else {
    valid_ = false;
J
jorlow@chromium.org 已提交
1352
  }
1353 1354 1355
  if (valid_ && prefix_same_as_start_) {
    assert(prefix_extractor_ != nullptr);
    prefix_.SetUserKey(prefix_extractor_->Transform(saved_key_.GetUserKey()));
1356
  }
J
jorlow@chromium.org 已提交
1357 1358
}

J
jorlow@chromium.org 已提交
1359
void DBIter::SeekToLast() {
Y
Yanqin Jin 已提交
1360 1361 1362 1363 1364 1365 1366
  if (timestamp_size_ > 0) {
    valid_ = false;
    status_ = Status::NotSupported(
        "SeekToLast/SeekForPrev/Prev currently not supported with timestamp.");
    return;
  }

1367 1368 1369
  if (iterate_upper_bound_ != nullptr) {
    // Seek to last key strictly less than ReadOptions.iterate_upper_bound.
    SeekForPrev(*iterate_upper_bound_);
1370
    if (Valid() && user_comparator_.Equal(*iterate_upper_bound_, key())) {
1371
      ReleaseTempPinnedData();
1372
      PrevInternal(nullptr);
1373 1374 1375 1376
    }
    return;
  }

1377
  PERF_CPU_TIMER_GUARD(iter_seek_cpu_nanos, env_);
S
Stanislau Hlebik 已提交
1378
  // Don't use iter_::Seek() if we set a prefix extractor
1379
  // because prefix seek will be used.
S
sdong 已提交
1380
  if (!expect_total_order_inner_iter()) {
S
Stanislau Hlebik 已提交
1381 1382
    max_skip_ = std::numeric_limits<uint64_t>::max();
  }
1383
  status_ = Status::OK();
J
jorlow@chromium.org 已提交
1384
  direction_ = kReverse;
1385
  ReleaseTempPinnedData();
1386
  ResetInternalKeysSkippedCounter();
J
jorlow@chromium.org 已提交
1387
  ClearSavedValue();
1388
  is_key_seqnum_zero_ = false;
1389 1390 1391

  {
    PERF_TIMER_GUARD(seek_internal_seek_time);
1392
    iter_.SeekToLast();
1393
    range_del_agg_.InvalidateRangeDelMapPositions();
1394
  }
1395
  PrevInternal(nullptr);
M
Manuel Ung 已提交
1396 1397 1398 1399 1400
  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());
1401
      PERF_COUNTER_ADD(iter_read_bytes, key().size() + value().size());
M
Manuel Ung 已提交
1402 1403
    }
  }
1404 1405 1406
  if (valid_ && prefix_same_as_start_) {
    assert(prefix_extractor_ != nullptr);
    prefix_.SetUserKey(prefix_extractor_->Transform(saved_key_.GetUserKey()));
1407
  }
J
jorlow@chromium.org 已提交
1408 1409
}

1410 1411
Iterator* NewDBIterator(Env* env, const ReadOptions& read_options,
                        const ImmutableCFOptions& cf_options,
1412
                        const MutableCFOptions& mutable_cf_options,
1413 1414 1415
                        const Comparator* user_key_comparator,
                        InternalIterator* internal_iter,
                        const SequenceNumber& sequence,
Y
Yi Wu 已提交
1416
                        uint64_t max_sequential_skip_in_iterations,
1417 1418 1419 1420 1421 1422
                        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);
1423
  return db_iter;
1424 1425
}

1426
}  // namespace ROCKSDB_NAMESPACE