git / moonbite-core / src/leveldb/table/table.cc · raw

// 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 "leveldb/table.h"#include "leveldb/cache.h"#include "leveldb/comparator.h"#include "leveldb/env.h"#include "leveldb/filter_policy.h"#include "leveldb/options.h"#include "table/block.h"#include "table/filter_block.h"#include "table/format.h"#include "table/two_level_iterator.h"#include "util/coding.h"namespace leveldb {struct Table::Rep {  ~Rep() {    delete filter;    delete[] filter_data;    delete index_block;  }  Options options;  Status status;  RandomAccessFile* file;  uint64_t cache_id;  FilterBlockReader* filter;  const char* filter_data;  BlockHandle metaindex_handle;  // Handle to metaindex_block: saved from footer  Block* index_block;};Status Table::Open(const Options& options, RandomAccessFile* file,                   uint64_t size, Table** table) {  *table = nullptr;  if (size < Footer::kEncodedLength) {    return Status::Corruption("file is too short to be an sstable");  }  char footer_space[Footer::kEncodedLength];  Slice footer_input;  Status s = file->Read(size - Footer::kEncodedLength, Footer::kEncodedLength,                        &footer_input, footer_space);  if (!s.ok()) return s;  Footer footer;  s = footer.DecodeFrom(&footer_input);  if (!s.ok()) return s;  // Read the index block  BlockContents index_block_contents;  if (s.ok()) {    ReadOptions opt;    if (options.paranoid_checks) {      opt.verify_checksums = true;    }    s = ReadBlock(file, opt, footer.index_handle(), &index_block_contents);  }  if (s.ok()) {    // We've successfully read the footer and the index block: we're    // ready to serve requests.    Block* index_block = new Block(index_block_contents);    Rep* rep = new Table::Rep;    rep->options = options;    rep->file = file;    rep->metaindex_handle = footer.metaindex_handle();    rep->index_block = index_block;    rep->cache_id = (options.block_cache ? options.block_cache->NewId() : 0);    rep->filter_data = nullptr;    rep->filter = nullptr;    *table = new Table(rep);    (*table)->ReadMeta(footer);  }  return s;}void Table::ReadMeta(const Footer& footer) {  if (rep_->options.filter_policy == nullptr) {    return;  // Do not need any metadata  }  // TODO(sanjay): Skip this if footer.metaindex_handle() size indicates  // it is an empty block.  ReadOptions opt;  if (rep_->options.paranoid_checks) {    opt.verify_checksums = true;  }  BlockContents contents;  if (!ReadBlock(rep_->file, opt, footer.metaindex_handle(), &contents).ok()) {    // Do not propagate errors since meta info is not needed for operation    return;  }  Block* meta = new Block(contents);  Iterator* iter = meta->NewIterator(BytewiseComparator());  std::string key = "filter.";  key.append(rep_->options.filter_policy->Name());  iter->Seek(key);  if (iter->Valid() && iter->key() == Slice(key)) {    ReadFilter(iter->value());  }  delete iter;  delete meta;}void Table::ReadFilter(const Slice& filter_handle_value) {  Slice v = filter_handle_value;  BlockHandle filter_handle;  if (!filter_handle.DecodeFrom(&v).ok()) {    return;  }  // We might want to unify with ReadBlock() if we start  // requiring checksum verification in Table::Open.  ReadOptions opt;  if (rep_->options.paranoid_checks) {    opt.verify_checksums = true;  }  BlockContents block;  if (!ReadBlock(rep_->file, opt, filter_handle, &block).ok()) {    return;  }  if (block.heap_allocated) {    rep_->filter_data = block.data.data();  // Will need to delete later  }  rep_->filter = new FilterBlockReader(rep_->options.filter_policy, block.data);}Table::~Table() { delete rep_; }static void DeleteBlock(void* arg, void* ignored) {  delete reinterpret_cast<Block*>(arg);}static void DeleteCachedBlock(const Slice& key, void* value) {  Block* block = reinterpret_cast<Block*>(value);  delete block;}static void ReleaseBlock(void* arg, void* h) {  Cache* cache = reinterpret_cast<Cache*>(arg);  Cache::Handle* handle = reinterpret_cast<Cache::Handle*>(h);  cache->Release(handle);}// Convert an index iterator value (i.e., an encoded BlockHandle)// into an iterator over the contents of the corresponding block.Iterator* Table::BlockReader(void* arg, const ReadOptions& options,                             const Slice& index_value) {  Table* table = reinterpret_cast<Table*>(arg);  Cache* block_cache = table->rep_->options.block_cache;  Block* block = nullptr;  Cache::Handle* cache_handle = nullptr;  BlockHandle handle;  Slice input = index_value;  Status s = handle.DecodeFrom(&input);  // We intentionally allow extra stuff in index_value so that we  // can add more features in the future.  if (s.ok()) {    BlockContents contents;    if (block_cache != nullptr) {      char cache_key_buffer[16];      EncodeFixed64(cache_key_buffer, table->rep_->cache_id);      EncodeFixed64(cache_key_buffer + 8, handle.offset());      Slice key(cache_key_buffer, sizeof(cache_key_buffer));      cache_handle = block_cache->Lookup(key);      if (cache_handle != nullptr) {        block = reinterpret_cast<Block*>(block_cache->Value(cache_handle));      } else {        s = ReadBlock(table->rep_->file, options, handle, &contents);        if (s.ok()) {          block = new Block(contents);          if (contents.cachable && options.fill_cache) {            cache_handle = block_cache->Insert(key, block, block->size(),                                               &DeleteCachedBlock);          }        }      }    } else {      s = ReadBlock(table->rep_->file, options, handle, &contents);      if (s.ok()) {        block = new Block(contents);      }    }  }  Iterator* iter;  if (block != nullptr) {    iter = block->NewIterator(table->rep_->options.comparator);    if (cache_handle == nullptr) {      iter->RegisterCleanup(&DeleteBlock, block, nullptr);    } else {      iter->RegisterCleanup(&ReleaseBlock, block_cache, cache_handle);    }  } else {    iter = NewErrorIterator(s);  }  return iter;}Iterator* Table::NewIterator(const ReadOptions& options) const {  return NewTwoLevelIterator(      rep_->index_block->NewIterator(rep_->options.comparator),      &Table::BlockReader, const_cast<Table*>(this), options);}Status Table::InternalGet(const ReadOptions& options, const Slice& k, void* arg,                          void (*handle_result)(void*, const Slice&,                                                const Slice&)) {  Status s;  Iterator* iiter = rep_->index_block->NewIterator(rep_->options.comparator);  iiter->Seek(k);  if (iiter->Valid()) {    Slice handle_value = iiter->value();    FilterBlockReader* filter = rep_->filter;    BlockHandle handle;    if (filter != nullptr && handle.DecodeFrom(&handle_value).ok() &&        !filter->KeyMayMatch(handle.offset(), k)) {      // Not found    } else {      Iterator* block_iter = BlockReader(this, options, iiter->value());      block_iter->Seek(k);      if (block_iter->Valid()) {        (*handle_result)(arg, block_iter->key(), block_iter->value());      }      s = block_iter->status();      delete block_iter;    }  }  if (s.ok()) {    s = iiter->status();  }  delete iiter;  return s;}uint64_t Table::ApproximateOffsetOf(const Slice& key) const {  Iterator* index_iter =      rep_->index_block->NewIterator(rep_->options.comparator);  index_iter->Seek(key);  uint64_t result;  if (index_iter->Valid()) {    BlockHandle handle;    Slice input = index_iter->value();    Status s = handle.DecodeFrom(&input);    if (s.ok()) {      result = handle.offset();    } else {      // Strange: we can't decode the block handle in the index block.      // We'll just return the offset of the metaindex block, which is      // close to the whole file size for this case.      result = rep_->metaindex_handle.offset();    }  } else {    // key is past the last key in the file.  Approximate the offset    // by returning the offset of the metaindex block (which is    // right near the end of the file).    result = rep_->metaindex_handle.offset();  }  delete index_iter;  return result;}}  // namespace leveldb