1 // Copyright (c) 2011-present, Facebook, Inc. All rights reserved.
2 // This source code is licensed under both the GPLv2 (found in the
3 // COPYING file in the root directory) and Apache 2.0 License
4 // (found in the LICENSE.Apache file in the root directory).
5
6 #ifndef ROCKSDB_LITE
7 #include "db/db_impl/compacted_db_impl.h"
8
9 #include "db/db_impl/db_impl.h"
10 #include "db/version_set.h"
11 #include "table/get_context.h"
12 #include "util/cast_util.h"
13
14 namespace ROCKSDB_NAMESPACE {
15
16 extern void MarkKeyMayExist(void* arg);
17 extern bool SaveValue(void* arg, const ParsedInternalKey& parsed_key,
18 const Slice& v, bool hit_and_return);
19
CompactedDBImpl(const DBOptions & options,const std::string & dbname)20 CompactedDBImpl::CompactedDBImpl(const DBOptions& options,
21 const std::string& dbname)
22 : DBImpl(options, dbname, /*seq_per_batch*/ false, +/*batch_per_txn*/ true,
23 /*read_only*/ true),
24 cfd_(nullptr),
25 version_(nullptr),
26 user_comparator_(nullptr) {}
27
~CompactedDBImpl()28 CompactedDBImpl::~CompactedDBImpl() {
29 }
30
FindFile(const Slice & key)31 size_t CompactedDBImpl::FindFile(const Slice& key) {
32 size_t right = files_.num_files - 1;
33 auto cmp = [&](const FdWithKeyRange& f, const Slice& k) -> bool {
34 return user_comparator_->Compare(ExtractUserKey(f.largest_key), k) < 0;
35 };
36 return static_cast<size_t>(std::lower_bound(files_.files,
37 files_.files + right, key, cmp) - files_.files);
38 }
39
Get(const ReadOptions & options,ColumnFamilyHandle *,const Slice & key,PinnableSlice * value)40 Status CompactedDBImpl::Get(const ReadOptions& options, ColumnFamilyHandle*,
41 const Slice& key, PinnableSlice* value) {
42 GetContext get_context(user_comparator_, nullptr, nullptr, nullptr,
43 GetContext::kNotFound, key, value, nullptr, nullptr,
44 nullptr, true, nullptr, nullptr);
45 LookupKey lkey(key, kMaxSequenceNumber);
46 Status s = files_.files[FindFile(key)].fd.table_reader->Get(
47 options, lkey.internal_key(), &get_context, nullptr);
48 if (!s.ok() && !s.IsNotFound()) {
49 return s;
50 }
51 if (get_context.State() == GetContext::kFound) {
52 return Status::OK();
53 }
54 return Status::NotFound();
55 }
56
MultiGet(const ReadOptions & options,const std::vector<ColumnFamilyHandle * > &,const std::vector<Slice> & keys,std::vector<std::string> * values)57 std::vector<Status> CompactedDBImpl::MultiGet(const ReadOptions& options,
58 const std::vector<ColumnFamilyHandle*>&,
59 const std::vector<Slice>& keys, std::vector<std::string>* values) {
60 autovector<TableReader*, 16> reader_list;
61 for (const auto& key : keys) {
62 const FdWithKeyRange& f = files_.files[FindFile(key)];
63 if (user_comparator_->Compare(key, ExtractUserKey(f.smallest_key)) < 0) {
64 reader_list.push_back(nullptr);
65 } else {
66 LookupKey lkey(key, kMaxSequenceNumber);
67 f.fd.table_reader->Prepare(lkey.internal_key());
68 reader_list.push_back(f.fd.table_reader);
69 }
70 }
71 std::vector<Status> statuses(keys.size(), Status::NotFound());
72 values->resize(keys.size());
73 int idx = 0;
74 for (auto* r : reader_list) {
75 if (r != nullptr) {
76 PinnableSlice pinnable_val;
77 std::string& value = (*values)[idx];
78 GetContext get_context(user_comparator_, nullptr, nullptr, nullptr,
79 GetContext::kNotFound, keys[idx], &pinnable_val,
80 nullptr, nullptr, nullptr, true, nullptr, nullptr);
81 LookupKey lkey(keys[idx], kMaxSequenceNumber);
82 Status s = r->Get(options, lkey.internal_key(), &get_context, nullptr);
83 assert(static_cast<size_t>(idx) < statuses.size());
84 if (!s.ok() && !s.IsNotFound()) {
85 statuses[idx] = s;
86 } else {
87 value.assign(pinnable_val.data(), pinnable_val.size());
88 if (get_context.State() == GetContext::kFound) {
89 statuses[idx] = Status::OK();
90 }
91 }
92 }
93 ++idx;
94 }
95 return statuses;
96 }
97
Init(const Options & options)98 Status CompactedDBImpl::Init(const Options& options) {
99 SuperVersionContext sv_context(/* create_superversion */ true);
100 mutex_.Lock();
101 ColumnFamilyDescriptor cf(kDefaultColumnFamilyName,
102 ColumnFamilyOptions(options));
103 Status s = Recover({cf}, true /* read only */, false, true);
104 if (s.ok()) {
105 cfd_ = static_cast_with_check<ColumnFamilyHandleImpl>(DefaultColumnFamily())
106 ->cfd();
107 cfd_->InstallSuperVersion(&sv_context, &mutex_);
108 }
109 mutex_.Unlock();
110 sv_context.Clean();
111 if (!s.ok()) {
112 return s;
113 }
114 NewThreadStatusCfInfo(cfd_);
115 version_ = cfd_->GetSuperVersion()->current;
116 user_comparator_ = cfd_->user_comparator();
117 auto* vstorage = version_->storage_info();
118 if (vstorage->num_non_empty_levels() == 0) {
119 return Status::NotSupported("no file exists");
120 }
121 const LevelFilesBrief& l0 = vstorage->LevelFilesBrief(0);
122 // L0 should not have files
123 if (l0.num_files > 1) {
124 return Status::NotSupported("L0 contain more than 1 file");
125 }
126 if (l0.num_files == 1) {
127 if (vstorage->num_non_empty_levels() > 1) {
128 return Status::NotSupported("Both L0 and other level contain files");
129 }
130 files_ = l0;
131 return Status::OK();
132 }
133
134 for (int i = 1; i < vstorage->num_non_empty_levels() - 1; ++i) {
135 if (vstorage->LevelFilesBrief(i).num_files > 0) {
136 return Status::NotSupported("Other levels also contain files");
137 }
138 }
139
140 int level = vstorage->num_non_empty_levels() - 1;
141 if (vstorage->LevelFilesBrief(level).num_files > 0) {
142 files_ = vstorage->LevelFilesBrief(level);
143 return Status::OK();
144 }
145 return Status::NotSupported("no file exists");
146 }
147
Open(const Options & options,const std::string & dbname,DB ** dbptr)148 Status CompactedDBImpl::Open(const Options& options,
149 const std::string& dbname, DB** dbptr) {
150 *dbptr = nullptr;
151
152 if (options.max_open_files != -1) {
153 return Status::InvalidArgument("require max_open_files = -1");
154 }
155 if (options.merge_operator.get() != nullptr) {
156 return Status::InvalidArgument("merge operator is not supported");
157 }
158 DBOptions db_options(options);
159 std::unique_ptr<CompactedDBImpl> db(new CompactedDBImpl(db_options, dbname));
160 Status s = db->Init(options);
161 if (s.ok()) {
162 db->StartPeriodicWorkScheduler();
163 ROCKS_LOG_INFO(db->immutable_db_options_.info_log,
164 "Opened the db as fully compacted mode");
165 LogFlush(db->immutable_db_options_.info_log);
166 *dbptr = db.release();
167 }
168 return s;
169 }
170
171 } // namespace ROCKSDB_NAMESPACE
172 #endif // ROCKSDB_LITE
173