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