You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
rocksdb/table/plain_table_reader.cc

457 lines
14 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 "table/plain_table_reader.h"
#include <unordered_map>
#include <map>
#include "db/dbformat.h"
#include "rocksdb/cache.h"
#include "rocksdb/comparator.h"
#include "rocksdb/env.h"
#include "rocksdb/filter_policy.h"
#include "rocksdb/options.h"
#include "rocksdb/statistics.h"
#include "table/block.h"
#include "table/filter_block.h"
#include "table/format.h"
#include "table/two_level_iterator.h"
#include "util/coding.h"
#include "util/hash.h"
#include "util/histogram.h"
#include "util/murmurhash.h"
#include "util/perf_context_imp.h"
#include "util/stop_watch.h"
namespace std {
template<>
struct hash<rocksdb::Slice> {
public:
std::size_t operator()(rocksdb::Slice const& s) const {
return MurmurHash(s.data(), s.size(), 397);
}
};
class slice_comparator {
public:
bool operator()(rocksdb::Slice const& s1, rocksdb::Slice const& s2) {
return s1.compare(s2) < 0;
}
};
}
namespace rocksdb {
static uint32_t getBucketId(Slice const& s, size_t prefix_len,
uint32_t num_buckets) {
return MurmurHash(s.data(), prefix_len, 397) % num_buckets;
}
PlainTableReader::PlainTableReader(const EnvOptions& storage_options,
uint64_t file_size, int user_key_size,
int key_prefix_len, int bloom_bits_per_key,
double hash_table_ratio) :
hash_table_size_(0), soptions_(storage_options), file_size_(file_size),
user_key_size_(user_key_size), key_prefix_len_(key_prefix_len),
hash_table_ratio_(hash_table_ratio) {
if (bloom_bits_per_key > 0) {
filter_policy_ = NewBloomFilterPolicy(bloom_bits_per_key);
} else {
filter_policy_ = nullptr;
}
hash_table_ = nullptr;
data_start_offset_ = 0;
data_end_offset_ = file_size;
}
PlainTableReader::~PlainTableReader() {
if (hash_table_ != nullptr) {
delete[] hash_table_;
}
if (filter_policy_ != nullptr) {
delete filter_policy_;
}
}
Status PlainTableReader::Open(const Options& options,
const EnvOptions& soptions,
unique_ptr<RandomAccessFile> && file,
uint64_t file_size,
unique_ptr<TableReader>* table_reader,
const int user_key_size,
const int key_prefix_len,
const int bloom_num_bits,
double hash_table_ratio) {
assert(options.allow_mmap_reads);
if (file_size > 2147483646) {
return Status::NotSupported("File is too large for PlainTableReader!");
}
PlainTableReader* t = new PlainTableReader(soptions, file_size,
user_key_size,
key_prefix_len,
bloom_num_bits,
hash_table_ratio);
t->file_ = std::move(file);
t->options_ = options;
Status s = t->PopulateIndex(file_size);
if (!s.ok()) {
delete t;
return s;
}
table_reader->reset(t);
return s;
}
void PlainTableReader::SetupForCompaction() {
}
bool PlainTableReader::PrefixMayMatch(const Slice& internal_prefix) {
return true;
}
Iterator* PlainTableReader::NewIterator(const ReadOptions& options) {
return new PlainTableIterator(this);
}
Status PlainTableReader::PopulateIndex(uint64_t file_size) {
// Get mmapped memory to file_data_.
Status s = file_->Read(0, file_size_, &file_data_, nullptr);
if (!s.ok()) {
return s;
}
version_ = DecodeFixed32(file_data_.data());
version_ ^= 0x80000000;
assert(version_ == 1);
data_start_offset_ = 4;
data_end_offset_ = file_size;
Slice key_slice;
Slice key_prefix_slice;
Slice key_suffix_slice;
Slice value_slice;
Slice prev_key_prefix_slice;
uint32_t pos = data_start_offset_;
int key_index_within_prefix = 0;
bool first = true;
std::string prefix_sub_index;
HistogramImpl keys_per_prefix_hist;
// Need map to be ordered to make sure sub indexes generated
// are in order.
std::map<Slice, std::string, std::slice_comparator> prefix2map;
while (pos < file_size) {
uint32_t key_offset = pos;
status_ = Next(pos, &key_slice, &value_slice, pos);
key_prefix_slice = Slice(key_slice.data(), key_prefix_len_);
if (first || prev_key_prefix_slice != key_prefix_slice) {
if (!first) {
keys_per_prefix_hist.Add(key_index_within_prefix);
}
key_index_within_prefix = 0;
prev_key_prefix_slice = key_prefix_slice;
}
if (key_index_within_prefix++ % 8 == 0) {
// Add an index key for every 8 keys
std::string& prefix_index = prefix2map[key_prefix_slice];
PutFixed32(&prefix_index, key_offset);
}
first = false;
}
keys_per_prefix_hist.Add(key_index_within_prefix);
if (hash_table_ != nullptr) {
delete[] hash_table_;
}
std::vector<Slice> filter_entries(0); // for creating bloom filter;
if (filter_policy_ != nullptr) {
filter_entries.reserve(prefix2map.size());
}
double hash_table_size_multipier =
(hash_table_ratio_ > 1.0) ? 1.0 : 1.0 / hash_table_ratio_;
hash_table_size_ = prefix2map.size() * hash_table_size_multipier + 1;
hash_table_ = new uint32_t[hash_table_size_];
std::vector<std::string> hash2map(hash_table_size_);
size_t sub_index_size_needed = 0;
for (auto& p: prefix2map) {
auto& sub_index = hash2map[getBucketId(p.first, key_prefix_len_,
hash_table_size_)];
if (sub_index.length() > 0 || p.second.length() > kOffsetLen) {
if (sub_index.length() <= kOffsetLen) {
sub_index_size_needed += sub_index.length() + 4;
}
sub_index_size_needed += p.second.length();
}
sub_index.append(p.second);
if (filter_policy_ != nullptr) {
filter_entries.push_back(p.first);
}
}
sub_index_.clear();
Log(options_.info_log, "Reserving %zu bytes for sub index",
sub_index_size_needed);
sub_index_.reserve(sub_index_size_needed);
for (int i = 0; i < hash_table_size_; i++) {
uint32_t num_keys_for_bucket = hash2map[i].length() / kOffsetLen;
switch (num_keys_for_bucket) {
case 0:
// No key for bucket
hash_table_[i] = data_end_offset_;
break;
case 1:
// point directly to the file offset
hash_table_[i] = DecodeFixed32(hash2map[i].data());
break;
default:
// point to index block
hash_table_[i] = sub_index_.length() | kSubIndexMask;
PutFixed32(&sub_index_, num_keys_for_bucket);
sub_index_.append(hash2map[i]);
}
}
if (filter_policy_ != nullptr) {
filter_str_.clear();
filter_policy_->CreateFilter(&filter_entries[0], filter_entries.size(),
&filter_str_);
filter_slice_ = Slice(filter_str_.data(), filter_str_.size());
}
Log(options_.info_log, "hash table size: %d, suffix_map length %zu",
hash_table_size_, sub_index_.length());
Log(options_.info_log, "Number of Keys per prefix Histogram: %s",
keys_per_prefix_hist.ToString().c_str());
return Status::OK();
}
uint32_t PlainTableReader::GetOffset(const Slice& target,
bool& prefix_matched) {
prefix_matched = false;
int bucket = getBucketId(target, key_prefix_len_, hash_table_size_);
uint32_t bucket_value = hash_table_[bucket];
if (bucket_value == data_end_offset_) {
return data_end_offset_;
} else if ((bucket_value & kSubIndexMask) == 0) {
// point directly to the file
return bucket_value;
}
// point to sub-index, need to do a binary search
uint32_t low = 0;
uint64_t prefix_index_offset = bucket_value ^ kSubIndexMask;
uint32_t upper_bound = DecodeFixed32(sub_index_.data() + prefix_index_offset);
uint32_t high = upper_bound;
uint64_t base_offset = prefix_index_offset + 4;
Slice mid_key;
// The key is between [low, high). Do a binary search between it.
while (high - low > 1) {
uint32_t mid = (high + low) / 2;
const char* index_offset = sub_index_.data() + base_offset
+ kOffsetLen * mid;
uint32_t file_offset = DecodeFixed32(index_offset);
mid_key = Slice(file_data_.data() + file_offset, user_key_size_);
int cmp_result = options_.comparator->Compare(target, mid_key);
if (cmp_result > 0) {
low = mid;
} else {
if (cmp_result == 0) {
// Happen to have found the exact key or target is smaller than the
// first key after base_offset.
prefix_matched = true;
return file_offset;
} else {
high = mid;
}
}
}
// The key is between low and low+1 (if exists). Both of them can have the
// correct prefix. Need to rule out at least one, to avoid to miss the
// correct one.
uint32_t low_key_offset = DecodeFixed32(
sub_index_.data() + base_offset + kOffsetLen * low);
if (low + 1 < upper_bound) {
if (Slice(file_data_.data() + low_key_offset, key_prefix_len_)
== Slice(target.data(), key_prefix_len_)) {
prefix_matched = true;
} else {
prefix_matched = false;
return DecodeFixed32(
sub_index_.data() + base_offset + kOffsetLen * (low + 1));
}
} else {
prefix_matched = false;
}
return low_key_offset;
}
bool PlainTableReader::MayHavePrefix(const Slice& target_prefix) {
return filter_policy_ == nullptr
|| filter_policy_->KeyMayMatch(target_prefix, filter_slice_);
}
Status PlainTableReader::Next(uint32_t offset, Slice* key, Slice* value,
uint32_t& next_offset) {
if (offset == data_end_offset_) {
next_offset = data_end_offset_;
return Status::OK();
}
if (offset > data_end_offset_) {
return Status::Corruption("Offset is out of file size");
}
int internal_key_size = GetInternalKeyLength();
if (offset + internal_key_size >= data_end_offset_) {
return Status::Corruption("Un able to read the next key");
}
const char* key_ptr = file_data_.data() + offset;
*key = Slice(key_ptr, internal_key_size);
uint32_t value_size;
const char* value_ptr = GetVarint32Ptr(key_ptr + internal_key_size,
file_data_.data() + data_end_offset_,
&value_size);
if (value_ptr == nullptr) {
return Status::Corruption("Error reading value length.");
}
next_offset = offset + (value_ptr - key_ptr) + value_size;
if (next_offset > data_end_offset_) {
return Status::Corruption("Reach end of file when reading value");
}
*value = Slice(value_ptr, value_size);
return Status::OK();
}
Status PlainTableReader::Get(
const ReadOptions& ro, const Slice& target, void* arg,
bool (*saver)(void*, const Slice&, const Slice&, bool),
void (*mark_key_may_exist)(void*)) {
// Check bloom filter first.
if (!MayHavePrefix(Slice(target.data(), key_prefix_len_))) {
return Status::OK();
}
uint32_t offset;
bool prefix_match;
offset = GetOffset(target, prefix_match);
Slice found_key;
Slice found_value;
while (offset < data_end_offset_) {
Status s = Next(offset, &found_key, &found_value, offset);
if (!s.ok()) {
return s;
}
if (!prefix_match) {
// Need to verify prefix for the first key found if it is not yet
// checked.
if (!target.starts_with(Slice(found_key.data(), key_prefix_len_))) {
break;
}
prefix_match = true;
}
if (options_.comparator->Compare(found_key, target) >= 0
&& !(*saver)(arg, found_key, found_value, true)) {
break;
}
}
return Status::OK();
}
bool PlainTableReader::TEST_KeyInCache(const ReadOptions& options,
const Slice& key) {
return false;
}
uint64_t PlainTableReader::ApproximateOffsetOf(const Slice& key) {
return 0;
}
PlainTableIterator::PlainTableIterator(PlainTableReader* table) :
table_(table) {
SeekToFirst();
}
PlainTableIterator::~PlainTableIterator() {
}
bool PlainTableIterator::Valid() const {
return offset_ < table_->data_end_offset_
&& offset_ >= table_->data_start_offset_;
}
void PlainTableIterator::SeekToFirst() {
next_offset_ = table_->data_start_offset_;
Next();
}
void PlainTableIterator::SeekToLast() {
assert(false);
}
void PlainTableIterator::Seek(const Slice& target) {
if (!table_->MayHavePrefix(Slice(target.data(), table_->key_prefix_len_))) {
offset_ = next_offset_ = table_->data_end_offset_;
return;
}
bool prefix_match;
next_offset_ = table_->GetOffset(target, prefix_match);
if (next_offset_ < table_-> data_end_offset_) {
for (Next(); status_.ok() && Valid(); Next()) {
if (!prefix_match) {
// Need to verify the first key's prefix
if (!target.starts_with(Slice(key().data(), table_->key_prefix_len_))) {
offset_ = next_offset_ = table_->data_end_offset_;
break;
}
prefix_match = true;
}
if (table_->options_.comparator->Compare(key(), target) >= 0) {
break;
}
}
} else {
offset_ = table_->data_end_offset_;
}
}
void PlainTableIterator::Next() {
offset_ = next_offset_;
Slice tmp_slice;
status_ = table_->Next(next_offset_, &key_, &value_, next_offset_);
}
void PlainTableIterator::Prev() {
assert(false);
}
Slice PlainTableIterator::key() const {
return key_;
}
Slice PlainTableIterator::value() const {
return value_;
}
Status PlainTableIterator::status() const {
return status_;
}
} // namespace rocksdb