Du kannst nicht mehr als 25 Themen auswählen
Themen müssen entweder mit einem Buchstaben oder einer Ziffer beginnen. Sie können Bindestriche („-“) enthalten und bis zu 35 Zeichen lang sein.
397 Zeilen
11 KiB
397 Zeilen
11 KiB
// 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"
|
|
|
|
#include "db/filename.h"
|
|
#include "db/dbformat.h"
|
|
#include "leveldb/env.h"
|
|
#include "leveldb/iterator.h"
|
|
#include "port/port.h"
|
|
#include "util/logging.h"
|
|
#include "util/mutexlock.h"
|
|
|
|
namespace leveldb {
|
|
|
|
#if 0
|
|
static void DumpInternalIter(Iterator* iter) {
|
|
for (iter->SeekToFirst(); iter->Valid(); iter->Next()) {
|
|
ParsedInternalKey k;
|
|
if (!ParseInternalKey(iter->key(), &k)) {
|
|
fprintf(stderr, "Corrupt '%s'\n", EscapeString(iter->key()).c_str());
|
|
} else {
|
|
fprintf(stderr, "@ '%s'\n", k.DebugString().c_str());
|
|
}
|
|
}
|
|
}
|
|
#endif
|
|
|
|
namespace {
|
|
|
|
// Memtables and sstables that make the DB representation contain
|
|
// (userkey,seq,type) => uservalue entries. DBIter
|
|
// combines multiple entries for the same userkey found in the DB
|
|
// representation into a single entry while accounting for sequence
|
|
// numbers, deletion markers, overwrites, etc.
|
|
class DBIter: public Iterator {
|
|
public:
|
|
// Which direction is the iterator currently moving?
|
|
// (1) When moving forward, the internal iterator is positioned at
|
|
// the exact entry that yields this->key(), this->value()
|
|
// (2) When moving backwards, the internal iterator is positioned
|
|
// just before all entries whose user key == this->key().
|
|
enum Direction {
|
|
kForward,
|
|
kReverse
|
|
};
|
|
|
|
DBIter(const std::string* dbname, Env* env,
|
|
const Comparator* cmp, Iterator* iter, SequenceNumber s)
|
|
: dbname_(dbname),
|
|
env_(env),
|
|
user_comparator_(cmp),
|
|
iter_(iter),
|
|
sequence_(s),
|
|
large_(NULL),
|
|
direction_(kForward),
|
|
valid_(false) {
|
|
}
|
|
virtual ~DBIter() {
|
|
delete iter_;
|
|
delete large_;
|
|
}
|
|
virtual bool Valid() const { return valid_; }
|
|
virtual Slice key() const {
|
|
assert(valid_);
|
|
return (direction_ == kForward) ? ExtractUserKey(iter_->key()) : saved_key_;
|
|
}
|
|
virtual Slice value() const {
|
|
assert(valid_);
|
|
Slice raw_value = (direction_ == kForward) ? iter_->value() : saved_value_;
|
|
if (large_ == NULL) {
|
|
return raw_value;
|
|
} else {
|
|
MutexLock l(&large_->mutex);
|
|
if (!large_->produced) {
|
|
ReadIndirectValue(raw_value);
|
|
}
|
|
return large_->value;
|
|
}
|
|
}
|
|
virtual Status status() const {
|
|
if (status_.ok()) {
|
|
if (large_ != NULL && !large_->status.ok()) return large_->status;
|
|
return iter_->status();
|
|
} else {
|
|
return status_;
|
|
}
|
|
}
|
|
|
|
virtual void Next();
|
|
virtual void Prev();
|
|
virtual void Seek(const Slice& target);
|
|
virtual void SeekToFirst();
|
|
virtual void SeekToLast();
|
|
|
|
private:
|
|
struct Large {
|
|
port::Mutex mutex;
|
|
std::string value;
|
|
bool produced;
|
|
Status status;
|
|
};
|
|
|
|
void FindNextUserEntry(bool skipping, std::string* skip);
|
|
void FindPrevUserEntry();
|
|
bool ParseKey(ParsedInternalKey* key);
|
|
void ReadIndirectValue(Slice ref) const;
|
|
|
|
inline void SaveKey(const Slice& k, std::string* dst) {
|
|
dst->assign(k.data(), k.size());
|
|
}
|
|
|
|
inline void ForgetLargeValue() {
|
|
if (large_ != NULL) {
|
|
delete large_;
|
|
large_ = NULL;
|
|
}
|
|
}
|
|
|
|
inline void ClearSavedValue() {
|
|
if (saved_value_.capacity() > 1048576) {
|
|
std::string empty;
|
|
swap(empty, saved_value_);
|
|
} else {
|
|
saved_value_.clear();
|
|
}
|
|
}
|
|
|
|
const std::string* const dbname_;
|
|
Env* const env_;
|
|
const Comparator* const user_comparator_;
|
|
Iterator* const iter_;
|
|
SequenceNumber const sequence_;
|
|
|
|
Status status_;
|
|
std::string saved_key_; // == current key when direction_==kReverse
|
|
std::string saved_value_; // == current raw value when direction_==kReverse
|
|
Large* large_; // Non-NULL if value is an indirect reference
|
|
Direction direction_;
|
|
bool valid_;
|
|
|
|
// No copying allowed
|
|
DBIter(const DBIter&);
|
|
void operator=(const DBIter&);
|
|
};
|
|
|
|
inline bool DBIter::ParseKey(ParsedInternalKey* ikey) {
|
|
if (!ParseInternalKey(iter_->key(), ikey)) {
|
|
status_ = Status::Corruption("corrupted internal key in DBIter");
|
|
return false;
|
|
} else {
|
|
return true;
|
|
}
|
|
}
|
|
|
|
void DBIter::Next() {
|
|
assert(valid_);
|
|
ForgetLargeValue();
|
|
|
|
if (direction_ == kReverse) { // Switch directions?
|
|
direction_ = kForward;
|
|
// iter_ is pointing just before the entries for this->key(),
|
|
// so advance into the range of entries for this->key() and then
|
|
// use the normal skipping code below.
|
|
if (!iter_->Valid()) {
|
|
iter_->SeekToFirst();
|
|
} else {
|
|
iter_->Next();
|
|
}
|
|
if (!iter_->Valid()) {
|
|
valid_ = false;
|
|
saved_key_.clear();
|
|
return;
|
|
}
|
|
}
|
|
|
|
// Temporarily use saved_key_ as storage for key to skip.
|
|
std::string* skip = &saved_key_;
|
|
SaveKey(ExtractUserKey(iter_->key()), skip);
|
|
FindNextUserEntry(true, skip);
|
|
}
|
|
|
|
void DBIter::FindNextUserEntry(bool skipping, std::string* skip) {
|
|
// Loop until we hit an acceptable entry to yield
|
|
assert(iter_->Valid());
|
|
assert(direction_ == kForward);
|
|
assert(large_ == NULL);
|
|
do {
|
|
ParsedInternalKey ikey;
|
|
if (ParseKey(&ikey) && ikey.sequence <= sequence_) {
|
|
switch (ikey.type) {
|
|
case kTypeDeletion:
|
|
// Arrange to skip all upcoming entries for this key since
|
|
// they are hidden by this deletion.
|
|
SaveKey(ikey.user_key, skip);
|
|
skipping = true;
|
|
break;
|
|
case kTypeValue:
|
|
case kTypeLargeValueRef:
|
|
if (skipping &&
|
|
user_comparator_->Compare(ikey.user_key, *skip) <= 0) {
|
|
// Entry hidden
|
|
} else {
|
|
valid_ = true;
|
|
saved_key_.clear();
|
|
if (ikey.type == kTypeLargeValueRef) {
|
|
large_ = new Large;
|
|
large_->produced = false;
|
|
}
|
|
return;
|
|
}
|
|
break;
|
|
}
|
|
}
|
|
iter_->Next();
|
|
} while (iter_->Valid());
|
|
saved_key_.clear();
|
|
valid_ = false;
|
|
}
|
|
|
|
void DBIter::Prev() {
|
|
assert(valid_);
|
|
ForgetLargeValue();
|
|
|
|
if (direction_ == kForward) { // Switch directions?
|
|
// iter_ is pointing at the current entry. Scan backwards until
|
|
// the key changes so we can use the normal reverse scanning code.
|
|
assert(iter_->Valid()); // Otherwise valid_ would have been false
|
|
SaveKey(ExtractUserKey(iter_->key()), &saved_key_);
|
|
while (true) {
|
|
iter_->Prev();
|
|
if (!iter_->Valid()) {
|
|
valid_ = false;
|
|
saved_key_.clear();
|
|
ClearSavedValue();
|
|
return;
|
|
}
|
|
if (user_comparator_->Compare(ExtractUserKey(iter_->key()),
|
|
saved_key_) < 0) {
|
|
break;
|
|
}
|
|
}
|
|
direction_ = kReverse;
|
|
}
|
|
|
|
FindPrevUserEntry();
|
|
}
|
|
|
|
void DBIter::FindPrevUserEntry() {
|
|
assert(direction_ == kReverse);
|
|
assert(large_ == NULL);
|
|
|
|
ValueType value_type = kTypeDeletion;
|
|
if (iter_->Valid()) {
|
|
SaveKey(ExtractUserKey(iter_->key()), &saved_key_);
|
|
do {
|
|
ParsedInternalKey ikey;
|
|
if (ParseKey(&ikey) && ikey.sequence <= sequence_) {
|
|
if ((value_type != kTypeDeletion) &&
|
|
user_comparator_->Compare(ikey.user_key, saved_key_) < 0) {
|
|
// We encountered a non-deleted value in entries for previous keys,
|
|
break;
|
|
}
|
|
value_type = ikey.type;
|
|
if (value_type == kTypeDeletion) {
|
|
ClearSavedValue();
|
|
} else {
|
|
Slice raw_value = iter_->value();
|
|
if (saved_value_.capacity() > raw_value.size() + 1048576) {
|
|
std::string empty;
|
|
swap(empty, saved_value_);
|
|
}
|
|
saved_value_.assign(raw_value.data(), raw_value.size());
|
|
}
|
|
}
|
|
iter_->Prev();
|
|
} while (iter_->Valid());
|
|
}
|
|
|
|
if (value_type == kTypeDeletion) {
|
|
// End
|
|
valid_ = false;
|
|
saved_key_.clear();
|
|
ClearSavedValue();
|
|
direction_ = kForward;
|
|
} else {
|
|
valid_ = true;
|
|
if (value_type == kTypeLargeValueRef) {
|
|
large_ = new Large;
|
|
large_->produced = false;
|
|
}
|
|
}
|
|
}
|
|
|
|
void DBIter::Seek(const Slice& target) {
|
|
direction_ = kForward;
|
|
ForgetLargeValue();
|
|
ClearSavedValue();
|
|
saved_key_.clear();
|
|
AppendInternalKey(
|
|
&saved_key_, ParsedInternalKey(target, sequence_, kValueTypeForSeek));
|
|
iter_->Seek(saved_key_);
|
|
if (iter_->Valid()) {
|
|
FindNextUserEntry(false, &saved_key_ /* temporary storage */);
|
|
} else {
|
|
valid_ = false;
|
|
}
|
|
}
|
|
|
|
void DBIter::SeekToFirst() {
|
|
direction_ = kForward;
|
|
ForgetLargeValue();
|
|
ClearSavedValue();
|
|
iter_->SeekToFirst();
|
|
if (iter_->Valid()) {
|
|
FindNextUserEntry(false, &saved_key_ /* temporary storage */);
|
|
} else {
|
|
valid_ = false;
|
|
}
|
|
}
|
|
|
|
void DBIter::SeekToLast() {
|
|
direction_ = kReverse;
|
|
ForgetLargeValue();
|
|
ClearSavedValue();
|
|
iter_->SeekToLast();
|
|
FindPrevUserEntry();
|
|
}
|
|
|
|
void DBIter::ReadIndirectValue(Slice ref) const {
|
|
assert(!large_->produced);
|
|
large_->produced = true;
|
|
LargeValueRef large_ref;
|
|
if (ref.size() != LargeValueRef::ByteSize()) {
|
|
large_->status = Status::Corruption("malformed large value reference");
|
|
return;
|
|
}
|
|
memcpy(large_ref.data, ref.data(), LargeValueRef::ByteSize());
|
|
std::string fname = LargeValueFileName(*dbname_, large_ref);
|
|
RandomAccessFile* file;
|
|
Status s = env_->NewRandomAccessFile(fname, &file);
|
|
uint64_t file_size = 0;
|
|
if (s.ok()) {
|
|
s = env_->GetFileSize(fname, &file_size);
|
|
}
|
|
if (s.ok()) {
|
|
uint64_t value_size = large_ref.ValueSize();
|
|
large_->value.resize(value_size);
|
|
Slice result;
|
|
s = file->Read(0, file_size, &result,
|
|
const_cast<char*>(large_->value.data()));
|
|
if (s.ok()) {
|
|
if (result.size() == file_size) {
|
|
switch (large_ref.compression_type()) {
|
|
case kNoCompression: {
|
|
if (result.data() != large_->value.data()) {
|
|
large_->value.assign(result.data(), result.size());
|
|
}
|
|
break;
|
|
}
|
|
case kSnappyCompression: {
|
|
std::string uncompressed;
|
|
if (port::Snappy_Uncompress(result.data(), result.size(),
|
|
&uncompressed) &&
|
|
uncompressed.size() == large_ref.ValueSize()) {
|
|
swap(uncompressed, large_->value);
|
|
} else {
|
|
s = Status::Corruption(
|
|
"Unable to read entire compressed large value file");
|
|
}
|
|
}
|
|
}
|
|
} else {
|
|
s = Status::Corruption("Unable to read entire large value file");
|
|
}
|
|
}
|
|
delete file; // Ignore errors on closing
|
|
}
|
|
if (!s.ok()) {
|
|
large_->value.clear();
|
|
large_->status = s;
|
|
}
|
|
}
|
|
|
|
} // anonymous namespace
|
|
|
|
Iterator* NewDBIterator(
|
|
const std::string* dbname,
|
|
Env* env,
|
|
const Comparator* user_key_comparator,
|
|
Iterator* internal_iter,
|
|
const SequenceNumber& sequence) {
|
|
return new DBIter(dbname, env, user_key_comparator, internal_iter, sequence);
|
|
}
|
|
|
|
}
|