← Back to branch summary

~ubuntu-branches/ubuntu/quantal/ceph/quantal

« back to all changes in this revision

Viewing changes to src/leveldb/table/table.cc

  • Committer: Package Import Robot
  • Author(s): James Page
  • Date: 2012-07-16 09:56:24 UTC
  • mfrom: (0.3.11)
  • mto: This revision was merged to the branch mainline in revision 17.
  • Revision ID: package-import@ubuntu.com-20120716095624-azr2w4hbhei1rxmx
Tags: upstream-0.48
Import upstream version 0.48

Show diffs side-by-side

added added

removed removed

Lines of Context:
src/leveldb/table/table.cc
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
 
}  // namespace leveldb