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 #include "leveldb/table.h" 6 7 #include "leveldb/cache.h" 8 #include "leveldb/env.h" 9 #include "table/block.h" 10 #include "table/format.h" 11 #include "table/two_level_iterator.h" 12 #include "util/coding.h" 13 14 namespace leveldb { 15 16 struct Table::Rep { 17 ~Rep() { 18 delete index_block; 19 } 20 21 Options options; 22 Status status; 23 RandomAccessFile* file; 24 uint64_t cache_id; 25 26 BlockHandle metaindex_handle; // Handle to metaindex_block: saved from footer 27 Block* index_block; 28 }; 29 30 Status Table::Open(const Options& options, 31 RandomAccessFile* file, 32 uint64_t size, 33 Table** table) { 34 *table = NULL; 35 if (size < Footer::kEncodedLength) { 36 return Status::InvalidArgument("file is too short to be an sstable"); 37 } 38 39 char footer_space[Footer::kEncodedLength]; 40 Slice footer_input; 41 Status s = file->Read(size - Footer::kEncodedLength, Footer::kEncodedLength, 42 &footer_input, footer_space); 43 if (!s.ok()) return s; 44 45 Footer footer; 46 s = footer.DecodeFrom(&footer_input); 47 if (!s.ok()) return s; 48 49 // Read the index block 50 Block* index_block = NULL; 51 if (s.ok()) { 52 s = ReadBlock(file, ReadOptions(), footer.index_handle(), &index_block); 53 } 54 55 if (s.ok()) { 56 // We've successfully read the footer and the index block: we're 57 // ready to serve requests. 58 Rep* rep = new Table::Rep; 59 rep->options = options; 60 rep->file = file; 61 rep->metaindex_handle = footer.metaindex_handle(); 62 rep->index_block = index_block; 63 rep->cache_id = (options.block_cache ? options.block_cache->NewId() : 0); 64 *table = new Table(rep); 65 } else { 66 if (index_block) delete index_block; 67 } 68 69 return s; 70 } 71 72 Table::~Table() { 73 delete rep_; 74 } 75 76 static void DeleteBlock(void* arg, void* ignored) { 77 delete reinterpret_cast<Block*>(arg); 78 } 79 80 static void DeleteCachedBlock(const Slice& key, void* value) { 81 Block* block = reinterpret_cast<Block*>(value); 82 delete block; 83 } 84 85 static void ReleaseBlock(void* arg, void* h) { 86 Cache* cache = reinterpret_cast<Cache*>(arg); 87 Cache::Handle* handle = reinterpret_cast<Cache::Handle*>(h); 88 cache->Release(handle); 89 } 90 91 // Convert an index iterator value (i.e., an encoded BlockHandle) 92 // into an iterator over the contents of the corresponding block. 93 Iterator* Table::BlockReader(void* arg, 94 const ReadOptions& options, 95 const Slice& index_value) { 96 Table* table = reinterpret_cast<Table*>(arg); 97 Cache* block_cache = table->rep_->options.block_cache; 98 Block* block = NULL; 99 Cache::Handle* cache_handle = NULL; 100 101 BlockHandle handle; 102 Slice input = index_value; 103 Status s = handle.DecodeFrom(&input); 104 // We intentionally allow extra stuff in index_value so that we 105 // can add more features in the future. 106 107 if (s.ok()) { 108 if (block_cache != NULL) { 109 char cache_key_buffer[16]; 110 EncodeFixed64(cache_key_buffer, table->rep_->cache_id); 111 EncodeFixed64(cache_key_buffer+8, handle.offset()); 112 Slice key(cache_key_buffer, sizeof(cache_key_buffer)); 113 cache_handle = block_cache->Lookup(key); 114 if (cache_handle != NULL) { 115 block = reinterpret_cast<Block*>(block_cache->Value(cache_handle)); 116 } else { 117 s = ReadBlock(table->rep_->file, options, handle, &block); 118 if (s.ok() && options.fill_cache) { 119 cache_handle = block_cache->Insert( 120 key, block, block->size(), &DeleteCachedBlock); 121 } 122 } 123 } else { 124 s = ReadBlock(table->rep_->file, options, handle, &block); 125 } 126 } 127 128 Iterator* iter; 129 if (block != NULL) { 130 iter = block->NewIterator(table->rep_->options.comparator); 131 if (cache_handle == NULL) { 132 iter->RegisterCleanup(&DeleteBlock, block, NULL); 133 } else { 134 iter->RegisterCleanup(&ReleaseBlock, block_cache, cache_handle); 135 } 136 } else { 137 iter = NewErrorIterator(s); 138 } 139 return iter; 140 } 141 142 Iterator* Table::NewIterator(const ReadOptions& options) const { 143 return NewTwoLevelIterator( 144 rep_->index_block->NewIterator(rep_->options.comparator), 145 &Table::BlockReader, const_cast<Table*>(this), options); 146 } 147 148 uint64_t Table::ApproximateOffsetOf(const Slice& key) const { 149 Iterator* index_iter = 150 rep_->index_block->NewIterator(rep_->options.comparator); 151 index_iter->Seek(key); 152 uint64_t result; 153 if (index_iter->Valid()) { 154 BlockHandle handle; 155 Slice input = index_iter->value(); 156 Status s = handle.DecodeFrom(&input); 157 if (s.ok()) { 158 result = handle.offset(); 159 } else { 160 // Strange: we can't decode the block handle in the index block. 161 // We'll just return the offset of the metaindex block, which is 162 // close to the whole file size for this case. 163 result = rep_->metaindex_handle.offset(); 164 } 165 } else { 166 // key is past the last key in the file. Approximate the offset 167 // by returning the offset of the metaindex block (which is 168 // right near the end of the file). 169 result = rep_->metaindex_handle.offset(); 170 } 171 delete index_iter; 172 return result; 173 } 174 175 } 176