write_batch.cc raw

   1  // Copyright (c) 2011 The LevelDB Authors. All rights reserved.
   2  // Use of this source code is governed by a BSD-style license that can be
   3  // found in the LICENSE file. See the AUTHORS file for names of contributors.
   4  //
   5  // WriteBatch::rep_ :=
   6  //    sequence: fixed64
   7  //    count: fixed32
   8  //    data: record[count]
   9  // record :=
  10  //    kTypeValue varstring varstring         |
  11  //    kTypeDeletion varstring
  12  // varstring :=
  13  //    len: varint32
  14  //    data: uint8[len]
  15  
  16  #include "leveldb/write_batch.h"
  17  
  18  #include "db/dbformat.h"
  19  #include "db/memtable.h"
  20  #include "db/write_batch_internal.h"
  21  #include "leveldb/db.h"
  22  #include "util/coding.h"
  23  
  24  namespace leveldb {
  25  
  26  // WriteBatch header has an 8-byte sequence number followed by a 4-byte count.
  27  static const size_t kHeader = 12;
  28  
  29  WriteBatch::WriteBatch() { Clear(); }
  30  
  31  WriteBatch::~WriteBatch() = default;
  32  
  33  WriteBatch::Handler::~Handler() = default;
  34  
  35  void WriteBatch::Clear() {
  36    rep_.clear();
  37    rep_.resize(kHeader);
  38  }
  39  
  40  size_t WriteBatch::ApproximateSize() const { return rep_.size(); }
  41  
  42  Status WriteBatch::Iterate(Handler* handler) const {
  43    Slice input(rep_);
  44    if (input.size() < kHeader) {
  45      return Status::Corruption("malformed WriteBatch (too small)");
  46    }
  47  
  48    input.remove_prefix(kHeader);
  49    Slice key, value;
  50    int found = 0;
  51    while (!input.empty()) {
  52      found++;
  53      char tag = input[0];
  54      input.remove_prefix(1);
  55      switch (tag) {
  56        case kTypeValue:
  57          if (GetLengthPrefixedSlice(&input, &key) &&
  58              GetLengthPrefixedSlice(&input, &value)) {
  59            handler->Put(key, value);
  60          } else {
  61            return Status::Corruption("bad WriteBatch Put");
  62          }
  63          break;
  64        case kTypeDeletion:
  65          if (GetLengthPrefixedSlice(&input, &key)) {
  66            handler->Delete(key);
  67          } else {
  68            return Status::Corruption("bad WriteBatch Delete");
  69          }
  70          break;
  71        default:
  72          return Status::Corruption("unknown WriteBatch tag");
  73      }
  74    }
  75    if (found != WriteBatchInternal::Count(this)) {
  76      return Status::Corruption("WriteBatch has wrong count");
  77    } else {
  78      return Status::OK();
  79    }
  80  }
  81  
  82  int WriteBatchInternal::Count(const WriteBatch* b) {
  83    return DecodeFixed32(b->rep_.data() + 8);
  84  }
  85  
  86  void WriteBatchInternal::SetCount(WriteBatch* b, int n) {
  87    EncodeFixed32(&b->rep_[8], n);
  88  }
  89  
  90  SequenceNumber WriteBatchInternal::Sequence(const WriteBatch* b) {
  91    return SequenceNumber(DecodeFixed64(b->rep_.data()));
  92  }
  93  
  94  void WriteBatchInternal::SetSequence(WriteBatch* b, SequenceNumber seq) {
  95    EncodeFixed64(&b->rep_[0], seq);
  96  }
  97  
  98  void WriteBatch::Put(const Slice& key, const Slice& value) {
  99    WriteBatchInternal::SetCount(this, WriteBatchInternal::Count(this) + 1);
 100    rep_.push_back(static_cast<char>(kTypeValue));
 101    PutLengthPrefixedSlice(&rep_, key);
 102    PutLengthPrefixedSlice(&rep_, value);
 103  }
 104  
 105  void WriteBatch::Delete(const Slice& key) {
 106    WriteBatchInternal::SetCount(this, WriteBatchInternal::Count(this) + 1);
 107    rep_.push_back(static_cast<char>(kTypeDeletion));
 108    PutLengthPrefixedSlice(&rep_, key);
 109  }
 110  
 111  void WriteBatch::Append(const WriteBatch& source) {
 112    WriteBatchInternal::Append(this, &source);
 113  }
 114  
 115  namespace {
 116  class MemTableInserter : public WriteBatch::Handler {
 117   public:
 118    SequenceNumber sequence_;
 119    MemTable* mem_;
 120  
 121    void Put(const Slice& key, const Slice& value) override {
 122      mem_->Add(sequence_, kTypeValue, key, value);
 123      sequence_++;
 124    }
 125    void Delete(const Slice& key) override {
 126      mem_->Add(sequence_, kTypeDeletion, key, Slice());
 127      sequence_++;
 128    }
 129  };
 130  }  // namespace
 131  
 132  Status WriteBatchInternal::InsertInto(const WriteBatch* b, MemTable* memtable) {
 133    MemTableInserter inserter;
 134    inserter.sequence_ = WriteBatchInternal::Sequence(b);
 135    inserter.mem_ = memtable;
 136    return b->Iterate(&inserter);
 137  }
 138  
 139  void WriteBatchInternal::SetContents(WriteBatch* b, const Slice& contents) {
 140    assert(contents.size() >= kHeader);
 141    b->rep_.assign(contents.data(), contents.size());
 142  }
 143  
 144  void WriteBatchInternal::Append(WriteBatch* dst, const WriteBatch* src) {
 145    SetCount(dst, Count(dst) + Count(src));
 146    assert(src->rep_.size() >= kHeader);
 147    dst->rep_.append(src->rep_.data() + kHeader, src->rep_.size() - kHeader);
 148  }
 149  
 150  }  // namespace leveldb
 151