memtable.cc 25.7 KB
Newer Older
1 2 3 4 5
//  Copyright (c) 2013, Facebook, Inc.  All rights reserved.
//  This source code is licensed under the BSD-style license found in the
//  LICENSE file in the root directory of this source tree. An additional grant
//  of patent rights can be found in the PATENTS file in the same directory.
//
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/memtable.h"
J
Jim Paton 已提交
11 12

#include <memory>
13
#include <algorithm>
14
#include <limits>
J
Jim Paton 已提交
15

J
jorlow@chromium.org 已提交
16
#include "db/dbformat.h"
17
#include "db/merge_context.h"
18
#include "db/writebuffer.h"
19 20 21 22
#include "rocksdb/comparator.h"
#include "rocksdb/env.h"
#include "rocksdb/iterator.h"
#include "rocksdb/merge_operator.h"
K
kailiu 已提交
23
#include "rocksdb/slice_transform.h"
24
#include "table/merger.h"
K
kailiu 已提交
25
#include "util/arena.h"
J
jorlow@chromium.org 已提交
26
#include "util/coding.h"
J
Jim Paton 已提交
27
#include "util/murmurhash.h"
K
Kai Liu 已提交
28
#include "util/mutexlock.h"
29
#include "util/perf_context_imp.h"
I
Igor Canadi 已提交
30
#include "util/statistics.h"
31
#include "util/stop_watch.h"
J
jorlow@chromium.org 已提交
32

33
namespace rocksdb {
J
jorlow@chromium.org 已提交
34

35
MemTableOptions::MemTableOptions(
L
Lei Jin 已提交
36 37
    const ImmutableCFOptions& ioptions,
    const MutableCFOptions& mutable_cf_options)
38 39 40 41 42
  : write_buffer_size(mutable_cf_options.write_buffer_size),
    arena_block_size(mutable_cf_options.arena_block_size),
    memtable_prefix_bloom_bits(mutable_cf_options.memtable_prefix_bloom_bits),
    memtable_prefix_bloom_probes(
        mutable_cf_options.memtable_prefix_bloom_probes),
L
Lei Jin 已提交
43
    memtable_prefix_bloom_huge_page_tlb_size(
44
        mutable_cf_options.memtable_prefix_bloom_huge_page_tlb_size),
L
Lei Jin 已提交
45 46 47
    inplace_update_support(ioptions.inplace_update_support),
    inplace_update_num_locks(mutable_cf_options.inplace_update_num_locks),
    inplace_callback(ioptions.inplace_callback),
48
    max_successive_merges(mutable_cf_options.max_successive_merges),
L
Lei Jin 已提交
49 50 51 52
    filter_deletes(mutable_cf_options.filter_deletes),
    statistics(ioptions.statistics),
    merge_operator(ioptions.merge_operator),
    info_log(ioptions.info_log) {}
L
Lei Jin 已提交
53 54 55

MemTable::MemTable(const InternalKeyComparator& cmp,
                   const ImmutableCFOptions& ioptions,
56
                   const MutableCFOptions& mutable_cf_options,
A
agiardullo 已提交
57
                   WriteBuffer* write_buffer, SequenceNumber earliest_seq)
J
jorlow@chromium.org 已提交
58
    : comparator_(cmp),
L
Lei Jin 已提交
59
      moptions_(ioptions, mutable_cf_options),
60
      refs_(0),
L
Lei Jin 已提交
61 62
      kArenaBlockSize(OptimizeBlockSize(moptions_.arena_block_size)),
      arena_(moptions_.arena_block_size),
63
      allocator_(&arena_, write_buffer),
L
Lei Jin 已提交
64
      table_(ioptions.memtable_factory->CreateMemTableRep(
65 66
          comparator_, &allocator_, ioptions.prefix_extractor,
          ioptions.info_log)),
67
      data_size_(0),
68
      num_entries_(0),
69
      num_deletes_(0),
70 71
      flush_in_progress_(false),
      flush_completed_(false),
A
Abhishek Kona 已提交
72
      file_number_(0),
73
      first_seqno_(0),
A
agiardullo 已提交
74
      earliest_seqno_(earliest_seq),
75
      mem_next_logfile_number_(0),
76 77 78
      locks_(moptions_.inplace_update_support
                 ? moptions_.inplace_update_num_locks
                 : 0),
L
Lei Jin 已提交
79
      prefix_extractor_(ioptions.prefix_extractor),
I
Igor Canadi 已提交
80
      should_flush_(ShouldFlushNow()),
81 82
      flush_scheduled_(false),
      env_(ioptions.env) {
83 84 85
  // if should_flush_ == true without an entry inserted, something must have
  // gone wrong already.
  assert(!should_flush_);
L
Lei Jin 已提交
86
  if (prefix_extractor_ && moptions_.memtable_prefix_bloom_bits > 0) {
87
    prefix_bloom_.reset(new DynamicBloom(
88
        &allocator_,
L
Lei Jin 已提交
89 90 91
        moptions_.memtable_prefix_bloom_bits, ioptions.bloom_locality,
        moptions_.memtable_prefix_bloom_probes, nullptr,
        moptions_.memtable_prefix_bloom_huge_page_tlb_size,
L
Lei Jin 已提交
92
        ioptions.info_log));
93 94
  }
}
J
jorlow@chromium.org 已提交
95

I
Igor Canadi 已提交
96
MemTable::~MemTable() { assert(refs_ == 0); }
J
jorlow@chromium.org 已提交
97

J
Jim Paton 已提交
98
size_t MemTable::ApproximateMemoryUsage() {
99 100 101 102 103 104 105 106 107 108
  size_t arena_usage = arena_.ApproximateMemoryUsage();
  size_t table_usage = table_->ApproximateMemoryUsage();
  // let MAX_USAGE =  std::numeric_limits<size_t>::max()
  // then if arena_usage + total_usage >= MAX_USAGE, return MAX_USAGE.
  // the following variation is to avoid numeric overflow.
  if (arena_usage >= std::numeric_limits<size_t>::max() - table_usage) {
    return std::numeric_limits<size_t>::max();
  }
  // otherwise, return the actual usage
  return arena_usage + table_usage;
J
Jim Paton 已提交
109
}
J
jorlow@chromium.org 已提交
110

111 112 113 114 115 116 117 118 119 120 121 122 123 124
bool MemTable::ShouldFlushNow() const {
  // In a lot of times, we cannot allocate arena blocks that exactly matches the
  // buffer size. Thus we have to decide if we should over-allocate or
  // under-allocate.
  // This constant avariable can be interpreted as: if we still have more than
  // "kAllowOverAllocationRatio * kArenaBlockSize" space left, we'd try to over
  // allocate one more block.
  const double kAllowOverAllocationRatio = 0.6;

  // If arena still have room for new block allocation, we can safely say it
  // shouldn't flush.
  auto allocated_memory =
      table_->ApproximateMemoryUsage() + arena_.MemoryAllocatedBytes();

125 126 127
  // if we can still allocate one more block without exceeding the
  // over-allocation ratio, then we should not flush.
  if (allocated_memory + kArenaBlockSize <
L
Lei Jin 已提交
128 129
      moptions_.write_buffer_size +
      kArenaBlockSize * kAllowOverAllocationRatio) {
130 131 132
    return false;
  }

L
Lei Jin 已提交
133 134 135 136 137
  // if user keeps adding entries that exceeds moptions.write_buffer_size,
  // we need to flush earlier even though we still have much available
  // memory left.
  if (allocated_memory > moptions_.write_buffer_size +
      kArenaBlockSize * kAllowOverAllocationRatio) {
138 139 140 141 142
    return true;
  }

  // In this code path, Arena has already allocated its "last block", which
  // means the total allocatedmemory size is either:
143
  //  (1) "moderately" over allocated the memory (no more than `0.6 * arena
144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168
  // block size`. Or,
  //  (2) the allocated memory is less than write buffer size, but we'll stop
  // here since if we allocate a new arena block, we'll over allocate too much
  // more (half of the arena block size) memory.
  //
  // In either case, to avoid over-allocate, the last block will stop allocation
  // when its usage reaches a certain ratio, which we carefully choose "0.75
  // full" as the stop condition because it addresses the following issue with
  // great simplicity: What if the next inserted entry's size is
  // bigger than AllocatedAndUnused()?
  //
  // The answer is: if the entry size is also bigger than 0.25 *
  // kArenaBlockSize, a dedicated block will be allocated for it; otherwise
  // arena will anyway skip the AllocatedAndUnused() and allocate a new, empty
  // and regular block. In either case, we *overly* over-allocated.
  //
  // Therefore, setting the last block to be at most "0.75 full" avoids both
  // cases.
  //
  // NOTE: the average percentage of waste space of this approach can be counted
  // as: "arena block size * 0.25 / write buffer size". User who specify a small
  // write buffer size and/or big arena block size may suffer.
  return arena_.AllocatedAndUnused() < kArenaBlockSize / 4;
}

169 170 171 172 173 174 175 176 177 178
int MemTable::KeyComparator::operator()(const char* prefix_len_key1,
                                        const char* prefix_len_key2) const {
  // Internal keys are encoded as length-prefixed strings.
  Slice k1 = GetLengthPrefixedSlice(prefix_len_key1);
  Slice k2 = GetLengthPrefixedSlice(prefix_len_key2);
  return comparator.Compare(k1, k2);
}

int MemTable::KeyComparator::operator()(const char* prefix_len_key,
                                        const Slice& key)
J
jorlow@chromium.org 已提交
179 180
    const {
  // Internal keys are encoded as length-prefixed strings.
181 182
  Slice a = GetLengthPrefixedSlice(prefix_len_key);
  return comparator.Compare(a, key);
J
jorlow@chromium.org 已提交
183 184
}

J
Jim Paton 已提交
185 186 187 188 189
Slice MemTableRep::UserKey(const char* key) const {
  Slice slice = GetLengthPrefixedSlice(key);
  return Slice(slice.data(), slice.size() - 8);
}

190
KeyHandle MemTableRep::Allocate(const size_t len, char** buf) {
191
  *buf = allocator_->Allocate(len);
192 193 194
  return static_cast<KeyHandle>(*buf);
}

J
jorlow@chromium.org 已提交
195 196 197
// Encode a suitable internal key target for "target" and return it.
// Uses *scratch as scratch space, and the returned pointer will point
// into this scratch space.
198
const char* EncodeKey(std::string* scratch, const Slice& target) {
J
jorlow@chromium.org 已提交
199
  scratch->clear();
200
  PutVarint32(scratch, static_cast<uint32_t>(target.size()));
J
jorlow@chromium.org 已提交
201 202 203 204 205 206
  scratch->append(target.data(), target.size());
  return scratch->data();
}

class MemTableIterator: public Iterator {
 public:
207
  MemTableIterator(
L
Lei Jin 已提交
208
      const MemTable& mem, const ReadOptions& read_options, Arena* arena)
209 210
      : bloom_(nullptr),
        prefix_extractor_(mem.prefix_extractor_),
211 212
        valid_(false),
        arena_mode_(arena != nullptr) {
L
Lei Jin 已提交
213
    if (prefix_extractor_ != nullptr && !read_options.total_order_seek) {
214
      bloom_ = mem.prefix_bloom_.get();
215
      iter_ = mem.table_->GetDynamicPrefixIterator(arena);
216
    } else {
217 218 219 220 221 222 223 224 225
      iter_ = mem.table_->GetIterator(arena);
    }
  }

  ~MemTableIterator() {
    if (arena_mode_) {
      iter_->~Iterator();
    } else {
      delete iter_;
226 227
    }
  }
J
Jim Paton 已提交
228

I
Igor Sugak 已提交
229 230
  virtual bool Valid() const override { return valid_; }
  virtual void Seek(const Slice& k) override {
I
Igor Canadi 已提交
231 232
    PERF_TIMER_GUARD(seek_on_memtable_time);
    PERF_COUNTER_ADD(seek_on_memtable_count, 1);
233 234
    if (bloom_ != nullptr &&
        !bloom_->MayContain(prefix_extractor_->Transform(ExtractUserKey(k)))) {
235 236 237 238 239 240
      valid_ = false;
      return;
    }
    iter_->Seek(k, nullptr);
    valid_ = iter_->Valid();
  }
I
Igor Sugak 已提交
241
  virtual void SeekToFirst() override {
242 243 244
    iter_->SeekToFirst();
    valid_ = iter_->Valid();
  }
I
Igor Sugak 已提交
245
  virtual void SeekToLast() override {
246 247 248
    iter_->SeekToLast();
    valid_ = iter_->Valid();
  }
I
Igor Sugak 已提交
249
  virtual void Next() override {
250 251 252 253
    assert(Valid());
    iter_->Next();
    valid_ = iter_->Valid();
  }
I
Igor Sugak 已提交
254
  virtual void Prev() override {
255 256 257 258
    assert(Valid());
    iter_->Prev();
    valid_ = iter_->Valid();
  }
I
Igor Sugak 已提交
259
  virtual Slice key() const override {
260
    assert(Valid());
J
Jim Paton 已提交
261 262
    return GetLengthPrefixedSlice(iter_->key());
  }
I
Igor Sugak 已提交
263
  virtual Slice value() const override {
264
    assert(Valid());
J
Jim Paton 已提交
265
    Slice key_slice = GetLengthPrefixedSlice(iter_->key());
J
jorlow@chromium.org 已提交
266 267 268
    return GetLengthPrefixedSlice(key_slice.data() + key_slice.size());
  }

I
Igor Sugak 已提交
269
  virtual Status status() const override { return Status::OK(); }
J
jorlow@chromium.org 已提交
270 271

 private:
272 273
  DynamicBloom* bloom_;
  const SliceTransform* const prefix_extractor_;
274
  MemTableRep::Iterator* iter_;
275
  bool valid_;
276
  bool arena_mode_;
J
jorlow@chromium.org 已提交
277 278 279 280 281 282

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

L
Lei Jin 已提交
283
Iterator* MemTable::NewIterator(const ReadOptions& read_options, Arena* arena) {
284 285
  assert(arena != nullptr);
  auto mem = arena->AllocateAligned(sizeof(MemTableIterator));
L
Lei Jin 已提交
286
  return new (mem) MemTableIterator(*this, read_options, arena);
J
jorlow@chromium.org 已提交
287 288
}

289
port::RWMutex* MemTable::GetLock(const Slice& key) {
K
kailiu 已提交
290 291
  static murmur_hash hash;
  return &locks_[hash(key) % locks_.size()];
292 293
}

294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313
uint64_t MemTable::ApproximateSize(const Slice& start_ikey,
                                   const Slice& end_ikey) {
  uint64_t entry_count = table_->ApproximateNumEntries(start_ikey, end_ikey);
  if (entry_count == 0) {
    return 0;
  }
  uint64_t n = num_entries_.load(std::memory_order_relaxed);
  if (n == 0) {
    return 0;
  }
  if (entry_count > n) {
    // table_->ApproximateNumEntries() is just an estimate so it can be larger
    // than actual entries we have. Cap it to entries we have to limit the
    // inaccuracy.
    entry_count = n;
  }
  uint64_t data_size = data_size_.load(std::memory_order_relaxed);
  return entry_count * (data_size / n);
}

J
jorlow@chromium.org 已提交
314
void MemTable::Add(SequenceNumber s, ValueType type,
315
                   const Slice& key, /* user key */
J
jorlow@chromium.org 已提交
316 317 318 319 320 321
                   const Slice& value) {
  // Format of an entry is concatenation of:
  //  key_size     : varint32 of internal_key.size()
  //  key bytes    : char[internal_key.size()]
  //  value_size   : varint32 of value.size()
  //  value bytes  : char[value.size()]
322 323 324 325 326 327
  uint32_t key_size = static_cast<uint32_t>(key.size());
  uint32_t val_size = static_cast<uint32_t>(value.size());
  uint32_t internal_key_size = key_size + 8;
  const uint32_t encoded_len = VarintLength(internal_key_size) +
                               internal_key_size + VarintLength(val_size) +
                               val_size;
328 329 330
  char* buf = nullptr;
  KeyHandle handle = table_->Allocate(encoded_len, &buf);
  assert(buf != nullptr);
J
jorlow@chromium.org 已提交
331 332 333
  char* p = EncodeVarint32(buf, internal_key_size);
  memcpy(p, key.data(), key_size);
  p += key_size;
A
agiardullo 已提交
334 335
  uint64_t packed = PackSequenceAndType(s, type);
  EncodeFixed64(p, packed);
J
jorlow@chromium.org 已提交
336 337 338
  p += 8;
  p = EncodeVarint32(p, val_size);
  memcpy(p, value.data(), val_size);
I
Igor Canadi 已提交
339
  assert((unsigned)(p + val_size - buf) == (unsigned)encoded_len);
340
  table_->Insert(handle);
341 342 343 344
  num_entries_.store(num_entries_.load(std::memory_order_relaxed) + 1,
                     std::memory_order_relaxed);
  data_size_.store(data_size_.load(std::memory_order_relaxed) + encoded_len,
                   std::memory_order_relaxed);
345 346 347
  if (type == kTypeDeletion) {
    num_deletes_++;
  }
348

349 350 351 352 353
  if (prefix_bloom_) {
    assert(prefix_extractor_);
    prefix_bloom_->Add(prefix_extractor_->Transform(key));
  }

354 355 356 357
  // The first sequence number inserted into the memtable
  assert(first_seqno_ == 0 || s > first_seqno_);
  if (first_seqno_ == 0) {
    first_seqno_ = s;
A
agiardullo 已提交
358 359 360 361 362

    if (earliest_seqno_ == kMaxSequenceNumber) {
      earliest_seqno_ = first_seqno_;
    }
    assert(first_seqno_ >= earliest_seqno_);
363
  }
364 365

  should_flush_ = ShouldFlushNow();
J
jorlow@chromium.org 已提交
366 367
}

368 369 370 371 372 373 374 375 376
// Callback from MemTable::Get()
namespace {

struct Saver {
  Status* status;
  const LookupKey* key;
  bool* found_final_value;  // Is value set correctly? Used by KeyMayExist
  bool* merge_in_progress;
  std::string* value;
A
agiardullo 已提交
377
  SequenceNumber seq;
378 379 380 381 382 383 384
  const MergeOperator* merge_operator;
  // the merge operations encountered;
  MergeContext* merge_context;
  MemTable* mem;
  Logger* logger;
  Statistics* statistics;
  bool inplace_update_support;
385
  Env* env_;
386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410
};
}  // namespace

static bool SaveValue(void* arg, const char* entry) {
  Saver* s = reinterpret_cast<Saver*>(arg);
  MergeContext* merge_context = s->merge_context;
  const MergeOperator* merge_operator = s->merge_operator;

  assert(s != nullptr && merge_context != nullptr);

  // entry format is:
  //    klength  varint32
  //    userkey  char[klength-8]
  //    tag      uint64
  //    vlength  varint32
  //    value    char[vlength]
  // Check that it belongs to same user key.  We do not check the
  // sequence number since the Seek() call above should have skipped
  // all entries with overly large sequence numbers.
  uint32_t key_length;
  const char* key_ptr = GetVarint32Ptr(entry, entry + 5, &key_length);
  if (s->mem->GetInternalKeyComparator().user_comparator()->Compare(
          Slice(key_ptr, key_length - 8), s->key->user_key()) == 0) {
    // Correct user key
    const uint64_t tag = DecodeFixed64(key_ptr + key_length - 8);
A
agiardullo 已提交
411 412 413 414
    ValueType type;
    UnPackSequenceAndType(tag, &s->seq, &type);

    switch (type) {
415 416 417 418 419 420 421 422
      case kTypeValue: {
        if (s->inplace_update_support) {
          s->mem->GetLock(s->key->user_key())->ReadLock();
        }
        Slice v = GetLengthPrefixedSlice(key_ptr + key_length);
        *(s->status) = Status::OK();
        if (*(s->merge_in_progress)) {
          assert(merge_operator);
423 424 425 426 427 428 429 430 431 432 433
          bool merge_success = false;
          {
            StopWatchNano timer(s->env_, s->statistics != nullptr);
            PERF_TIMER_GUARD(merge_operator_time_nanos);
            merge_success = merge_operator->FullMerge(
                s->key->user_key(), &v, merge_context->GetOperands(), s->value,
                s->logger);
            RecordTick(s->statistics, MERGE_OPERATION_TOTAL_TIME,
                       timer.ElapsedNanos());
          }
          if (!merge_success) {
434 435 436 437 438 439 440 441
            RecordTick(s->statistics, NUMBER_MERGE_FAILURES);
            *(s->status) =
                Status::Corruption("Error: Could not perform merge.");
          }
        } else {
          s->value->assign(v.data(), v.size());
        }
        if (s->inplace_update_support) {
442
          s->mem->GetLock(s->key->user_key())->ReadUnlock();
443 444 445 446 447 448 449 450
        }
        *(s->found_final_value) = true;
        return false;
      }
      case kTypeDeletion: {
        if (*(s->merge_in_progress)) {
          assert(merge_operator);
          *(s->status) = Status::OK();
451 452 453 454 455 456 457 458 459 460 461
          bool merge_success = false;
          {
            StopWatchNano timer(s->env_, s->statistics != nullptr);
            PERF_TIMER_GUARD(merge_operator_time_nanos);
            merge_success = merge_operator->FullMerge(
                s->key->user_key(), nullptr, merge_context->GetOperands(),
                s->value, s->logger);
            RecordTick(s->statistics, MERGE_OPERATION_TOTAL_TIME,
                       timer.ElapsedNanos());
          }
          if (!merge_success) {
462 463 464 465 466 467 468 469 470 471 472
            RecordTick(s->statistics, NUMBER_MERGE_FAILURES);
            *(s->status) =
                Status::Corruption("Error: Could not perform merge.");
          }
        } else {
          *(s->status) = Status::NotFound();
        }
        *(s->found_final_value) = true;
        return false;
      }
      case kTypeMerge: {
473 474 475 476
        if (!merge_operator) {
          *(s->status) = Status::InvalidArgument(
              "merge_operator is not properly initialized.");
          // Normally we continue the loop (return true) when we see a merge
477
          // operand.  But in case of an error, we should stop the loop
478 479 480 481 482
          // immediately and pretend we have found the value to stop further
          // seek.  Otherwise, the later call will override this error status.
          *(s->found_final_value) = true;
          return false;
        }
483 484 485 486 487 488 489 490 491 492 493 494 495 496 497
        Slice v = GetLengthPrefixedSlice(key_ptr + key_length);
        *(s->merge_in_progress) = true;
        merge_context->PushOperand(v);
        return true;
      }
      default:
        assert(false);
        return true;
    }
  }

  // s->state could be Corrupt, merge or notfound
  return false;
}

498
bool MemTable::Get(const LookupKey& key, std::string* value, Status* s,
A
agiardullo 已提交
499
                   MergeContext* merge_context, SequenceNumber* seq) {
500
  // The sequence number is updated synchronously in version_set.h
501
  if (IsEmpty()) {
502 503 504
    // Avoiding recording stats for speed.
    return false;
  }
505
  PERF_TIMER_GUARD(get_from_memtable_time);
506

507
  Slice user_key = key.user_key();
508 509
  bool found_final_value = false;
  bool merge_in_progress = s->IsMergeInProgress();
510 511 512 513

  if (prefix_bloom_ &&
      !prefix_bloom_->MayContain(prefix_extractor_->Transform(user_key))) {
    // iter is null if prefix bloom says the key does not exist
A
agiardullo 已提交
514
    *seq = kMaxSequenceNumber;
515
  } else {
516 517 518 519 520 521
    Saver saver;
    saver.status = s;
    saver.found_final_value = &found_final_value;
    saver.merge_in_progress = &merge_in_progress;
    saver.key = &key;
    saver.value = value;
A
agiardullo 已提交
522
    saver.seq = kMaxSequenceNumber;
523
    saver.mem = this;
524
    saver.merge_context = merge_context;
L
Lei Jin 已提交
525 526
    saver.merge_operator = moptions_.merge_operator;
    saver.logger = moptions_.info_log;
L
Lei Jin 已提交
527
    saver.inplace_update_support = moptions_.inplace_update_support;
L
Lei Jin 已提交
528
    saver.statistics = moptions_.statistics;
529
    saver.env_ = env_;
530
    table_->Get(key, &saver, SaveValue);
A
agiardullo 已提交
531 532

    *seq = saver.seq;
533
  }
534

535
  // No change to value, since we have not yet found a Put/Delete
536
  if (!found_final_value && merge_in_progress) {
537 538
    *s = Status::MergeInProgress("");
  }
L
Lei Jin 已提交
539
  PERF_COUNTER_ADD(get_from_memtable_count, 1);
540
  return found_final_value;
541 542
}

543
void MemTable::Update(SequenceNumber seq,
544 545 546
                      const Slice& key,
                      const Slice& value) {
  LookupKey lkey(key, seq);
547
  Slice mem_key = lkey.memtable_key();
548

549
  std::unique_ptr<MemTableRep::Iterator> iter(
550
      table_->GetDynamicPrefixIterator());
551
  iter->Seek(lkey.internal_key(), mem_key.data());
552 553 554

  if (iter->Valid()) {
    // entry format is:
555
    //    key_length  varint32
556 557 558 559 560 561 562 563
    //    userkey  char[klength-8]
    //    tag      uint64
    //    vlength  varint32
    //    value    char[vlength]
    // Check that it belongs to same user key.  We do not check the
    // sequence number since the Seek() call above should have skipped
    // all entries with overly large sequence numbers.
    const char* entry = iter->key();
K
Kai Liu 已提交
564
    uint32_t key_length = 0;
565 566 567 568 569
    const char* key_ptr = GetVarint32Ptr(entry, entry + 5, &key_length);
    if (comparator_.comparator.user_comparator()->Compare(
        Slice(key_ptr, key_length - 8), lkey.user_key()) == 0) {
      // Correct user key
      const uint64_t tag = DecodeFixed64(key_ptr + key_length - 8);
A
agiardullo 已提交
570 571 572 573
      ValueType type;
      SequenceNumber unused;
      UnPackSequenceAndType(tag, &unused, &type);
      switch (type) {
574
        case kTypeValue: {
575
          Slice prev_value = GetLengthPrefixedSlice(key_ptr + key_length);
576 577
          uint32_t prev_size = static_cast<uint32_t>(prev_value.size());
          uint32_t new_size = static_cast<uint32_t>(value.size());
578

579 580
          // Update value, if new value size  <= previous value size
          if (new_size <= prev_size ) {
581
            char* p = EncodeVarint32(const_cast<char*>(key_ptr) + key_length,
582
                                     new_size);
583 584
            WriteLock wl(GetLock(lkey.user_key()));
            memcpy(p, value.data(), value.size());
I
Igor Canadi 已提交
585 586 587
            assert((unsigned)((p + value.size()) - entry) ==
                   (unsigned)(VarintLength(key_length) + key_length +
                              VarintLength(value.size()) + value.size()));
588
            return;
589 590 591 592
          }
        }
        default:
          // If the latest value is kTypeDeletion, kTypeMerge or kTypeLogData
593 594 595
          // we don't have enough space for update inplace
            Add(seq, kTypeValue, key, value);
            return;
596 597 598 599
      }
    }
  }

600 601 602 603 604
  // key doesn't exist
  Add(seq, kTypeValue, key, value);
}

bool MemTable::UpdateCallback(SequenceNumber seq,
605
                              const Slice& key,
L
Lei Jin 已提交
606
                              const Slice& delta) {
607 608 609
  LookupKey lkey(key, seq);
  Slice memkey = lkey.memtable_key();

L
Lei Jin 已提交
610
  std::unique_ptr<MemTableRep::Iterator> iter(
611
      table_->GetDynamicPrefixIterator());
612
  iter->Seek(lkey.internal_key(), memkey.data());
613 614 615 616 617 618 619 620 621 622 623 624

  if (iter->Valid()) {
    // entry format is:
    //    key_length  varint32
    //    userkey  char[klength-8]
    //    tag      uint64
    //    vlength  varint32
    //    value    char[vlength]
    // Check that it belongs to same user key.  We do not check the
    // sequence number since the Seek() call above should have skipped
    // all entries with overly large sequence numbers.
    const char* entry = iter->key();
K
Kai Liu 已提交
625
    uint32_t key_length = 0;
626 627 628 629 630
    const char* key_ptr = GetVarint32Ptr(entry, entry + 5, &key_length);
    if (comparator_.comparator.user_comparator()->Compare(
        Slice(key_ptr, key_length - 8), lkey.user_key()) == 0) {
      // Correct user key
      const uint64_t tag = DecodeFixed64(key_ptr + key_length - 8);
A
agiardullo 已提交
631 632 633 634
      ValueType type;
      uint64_t unused;
      UnPackSequenceAndType(tag, &unused, &type);
      switch (type) {
635 636
        case kTypeValue: {
          Slice prev_value = GetLengthPrefixedSlice(key_ptr + key_length);
637
          uint32_t prev_size = static_cast<uint32_t>(prev_value.size());
638 639

          char* prev_buffer = const_cast<char*>(prev_value.data());
640
          uint32_t new_prev_size = prev_size;
641 642

          std::string str_value;
643
          WriteLock wl(GetLock(lkey.user_key()));
L
Lei Jin 已提交
644 645
          auto status = moptions_.inplace_callback(prev_buffer, &new_prev_size,
                                                   delta, &str_value);
646
          if (status == UpdateStatus::UPDATED_INPLACE) {
647
            // Value already updated by callback.
648 649 650 651 652 653 654 655 656 657
            assert(new_prev_size <= prev_size);
            if (new_prev_size < prev_size) {
              // overwrite the new prev_size
              char* p = EncodeVarint32(const_cast<char*>(key_ptr) + key_length,
                                       new_prev_size);
              if (VarintLength(new_prev_size) < VarintLength(prev_size)) {
                // shift the value buffer as well.
                memcpy(p, prev_buffer, new_prev_size);
              }
            }
L
Lei Jin 已提交
658
            RecordTick(moptions_.statistics, NUMBER_KEYS_UPDATED);
659
            should_flush_ = ShouldFlushNow();
660
            return true;
661 662
          } else if (status == UpdateStatus::UPDATED) {
            Add(seq, kTypeValue, key, Slice(str_value));
L
Lei Jin 已提交
663
            RecordTick(moptions_.statistics, NUMBER_KEYS_WRITTEN);
664
            should_flush_ = ShouldFlushNow();
665
            return true;
666 667
          } else if (status == UpdateStatus::UPDATE_FAILED) {
            // No action required. Return.
668
            should_flush_ = ShouldFlushNow();
669 670 671 672 673 674 675 676 677 678
            return true;
          }
        }
        default:
          break;
      }
    }
  }
  // If the latest value is not kTypeValue
  // or key doesn't exist
679 680
  return false;
}
681 682 683 684 685 686 687

size_t MemTable::CountSuccessiveMergeEntries(const LookupKey& key) {
  Slice memkey = key.memtable_key();

  // A total ordered iterator is costly for some memtablerep (prefix aware
  // reps). By passing in the user key, we allow efficient iterator creation.
  // The iterator only needs to be ordered within the same user key.
688
  std::unique_ptr<MemTableRep::Iterator> iter(
689
      table_->GetDynamicPrefixIterator());
690
  iter->Seek(key.internal_key(), memkey.data());
691 692 693 694 695

  size_t num_successive_merges = 0;

  for (; iter->Valid(); iter->Next()) {
    const char* entry = iter->key();
K
Kai Liu 已提交
696
    uint32_t key_length = 0;
697
    const char* iter_key_ptr = GetVarint32Ptr(entry, entry + 5, &key_length);
I
Igor Canadi 已提交
698 699
    if (comparator_.comparator.user_comparator()->Compare(
            Slice(iter_key_ptr, key_length - 8), key.user_key()) != 0) {
700 701 702 703
      break;
    }

    const uint64_t tag = DecodeFixed64(iter_key_ptr + key_length - 8);
A
agiardullo 已提交
704 705 706 707
    ValueType type;
    uint64_t unused;
    UnPackSequenceAndType(tag, &unused, &type);
    if (type != kTypeMerge) {
708 709 710 711 712 713 714 715 716
      break;
    }

    ++num_successive_merges;
  }

  return num_successive_merges;
}

717 718
void MemTableRep::Get(const LookupKey& k, void* callback_args,
                      bool (*callback_func)(void* arg, const char* entry)) {
719
  auto iter = GetDynamicPrefixIterator();
720 721 722 723 724 725
  for (iter->Seek(k.internal_key(), k.memtable_key().data());
       iter->Valid() && callback_func(callback_args, iter->key());
       iter->Next()) {
  }
}

726
}  // namespace rocksdb