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 // Test for issue 178: a manual compaction causes deleted data to reappear.
7 #include <cstdlib>
8
9 #include "port/port.h"
10 #include "rocksdb/compaction_filter.h"
11 #include "rocksdb/db.h"
12 #include "rocksdb/slice.h"
13 #include "rocksdb/write_batch.h"
14 #include "test_util/testharness.h"
15
16 using namespace ROCKSDB_NAMESPACE;
17
18 namespace {
19
20 // Reasoning: previously the number was 1100000. Since the keys are written to
21 // the batch in one write each write will result into one SST file. each write
22 // will result into one SST file. We reduced the write_buffer_size to 1K to
23 // basically have the same effect with however less number of keys, which
24 // results into less test runtime.
25 const int kNumKeys = 1100;
26
Key1(int i)27 std::string Key1(int i) {
28 char buf[100];
29 snprintf(buf, sizeof(buf), "my_key_%d", i);
30 return buf;
31 }
32
Key2(int i)33 std::string Key2(int i) {
34 return Key1(i) + "_xxx";
35 }
36
37 class ManualCompactionTest : public testing::Test {
38 public:
ManualCompactionTest()39 ManualCompactionTest() {
40 // Get rid of any state from an old run.
41 dbname_ = test::PerThreadDBPath("rocksdb_manual_compaction_test");
42 DestroyDB(dbname_, Options());
43 }
44
45 std::string dbname_;
46 };
47
48 class DestroyAllCompactionFilter : public CompactionFilter {
49 public:
DestroyAllCompactionFilter()50 DestroyAllCompactionFilter() {}
51
Filter(int,const Slice &,const Slice & existing_value,std::string *,bool *) const52 bool Filter(int /*level*/, const Slice& /*key*/, const Slice& existing_value,
53 std::string* /*new_value*/,
54 bool* /*value_changed*/) const override {
55 return existing_value.ToString() == "destroy";
56 }
57
Name() const58 const char* Name() const override { return "DestroyAllCompactionFilter"; }
59 };
60
61 class LogCompactionFilter : public CompactionFilter {
62 public:
Name() const63 const char* Name() const override { return "LogCompactionFilter"; }
64
Filter(int level,const Slice & key,const Slice &,std::string *,bool *) const65 bool Filter(int level, const Slice& key, const Slice& /*existing_value*/,
66 std::string* /*new_value*/,
67 bool* /*value_changed*/) const override {
68 key_level_[key.ToString()] = level;
69 return false;
70 }
71
Reset()72 void Reset() { key_level_.clear(); }
73
NumKeys() const74 size_t NumKeys() const { return key_level_.size(); }
75
KeyLevel(const Slice & key)76 int KeyLevel(const Slice& key) {
77 auto it = key_level_.find(key.ToString());
78 if (it == key_level_.end()) {
79 return -1;
80 }
81 return it->second;
82 }
83
84 private:
85 mutable std::map<std::string, int> key_level_;
86 };
87
TEST_F(ManualCompactionTest,CompactTouchesAllKeys)88 TEST_F(ManualCompactionTest, CompactTouchesAllKeys) {
89 for (int iter = 0; iter < 2; ++iter) {
90 DB* db;
91 Options options;
92 if (iter == 0) { // level compaction
93 options.num_levels = 3;
94 options.compaction_style = kCompactionStyleLevel;
95 } else { // universal compaction
96 options.compaction_style = kCompactionStyleUniversal;
97 }
98 options.create_if_missing = true;
99 options.compression = kNoCompression;
100 options.compaction_filter = new DestroyAllCompactionFilter();
101 ASSERT_OK(DB::Open(options, dbname_, &db));
102
103 db->Put(WriteOptions(), Slice("key1"), Slice("destroy"));
104 db->Put(WriteOptions(), Slice("key2"), Slice("destroy"));
105 db->Put(WriteOptions(), Slice("key3"), Slice("value3"));
106 db->Put(WriteOptions(), Slice("key4"), Slice("destroy"));
107
108 Slice key4("key4");
109 db->CompactRange(CompactRangeOptions(), nullptr, &key4);
110 Iterator* itr = db->NewIterator(ReadOptions());
111 itr->SeekToFirst();
112 ASSERT_TRUE(itr->Valid());
113 ASSERT_EQ("key3", itr->key().ToString());
114 itr->Next();
115 ASSERT_TRUE(!itr->Valid());
116 delete itr;
117
118 delete options.compaction_filter;
119 delete db;
120 DestroyDB(dbname_, options);
121 }
122 }
123
TEST_F(ManualCompactionTest,Test)124 TEST_F(ManualCompactionTest, Test) {
125 // Open database. Disable compression since it affects the creation
126 // of layers and the code below is trying to test against a very
127 // specific scenario.
128 DB* db;
129 Options db_options;
130 db_options.write_buffer_size = 1024;
131 db_options.create_if_missing = true;
132 db_options.compression = kNoCompression;
133 ASSERT_OK(DB::Open(db_options, dbname_, &db));
134
135 // create first key range
136 WriteBatch batch;
137 for (int i = 0; i < kNumKeys; i++) {
138 batch.Put(Key1(i), "value for range 1 key");
139 }
140 ASSERT_OK(db->Write(WriteOptions(), &batch));
141
142 // create second key range
143 batch.Clear();
144 for (int i = 0; i < kNumKeys; i++) {
145 batch.Put(Key2(i), "value for range 2 key");
146 }
147 ASSERT_OK(db->Write(WriteOptions(), &batch));
148
149 // delete second key range
150 batch.Clear();
151 for (int i = 0; i < kNumKeys; i++) {
152 batch.Delete(Key2(i));
153 }
154 ASSERT_OK(db->Write(WriteOptions(), &batch));
155
156 // compact database
157 std::string start_key = Key1(0);
158 std::string end_key = Key1(kNumKeys - 1);
159 Slice least(start_key.data(), start_key.size());
160 Slice greatest(end_key.data(), end_key.size());
161
162 // commenting out the line below causes the example to work correctly
163 db->CompactRange(CompactRangeOptions(), &least, &greatest);
164
165 // count the keys
166 Iterator* iter = db->NewIterator(ReadOptions());
167 int num_keys = 0;
168 for (iter->SeekToFirst(); iter->Valid(); iter->Next()) {
169 num_keys++;
170 }
171 delete iter;
172 ASSERT_EQ(kNumKeys, num_keys) << "Bad number of keys";
173
174 // close database
175 delete db;
176 DestroyDB(dbname_, Options());
177 }
178
TEST_F(ManualCompactionTest,SkipLevel)179 TEST_F(ManualCompactionTest, SkipLevel) {
180 DB* db;
181 Options options;
182 options.num_levels = 3;
183 // Initially, flushed L0 files won't exceed 100.
184 options.level0_file_num_compaction_trigger = 100;
185 options.compaction_style = kCompactionStyleLevel;
186 options.create_if_missing = true;
187 options.compression = kNoCompression;
188 LogCompactionFilter* filter = new LogCompactionFilter();
189 options.compaction_filter = filter;
190 ASSERT_OK(DB::Open(options, dbname_, &db));
191
192 WriteOptions wo;
193 FlushOptions fo;
194 ASSERT_OK(db->Put(wo, "1", ""));
195 ASSERT_OK(db->Flush(fo));
196 ASSERT_OK(db->Put(wo, "2", ""));
197 ASSERT_OK(db->Flush(fo));
198 ASSERT_OK(db->Put(wo, "4", ""));
199 ASSERT_OK(db->Put(wo, "8", ""));
200 ASSERT_OK(db->Flush(fo));
201
202 {
203 // L0: 1, 2, [4, 8]
204 // no file has keys in range [5, 7]
205 Slice start("5");
206 Slice end("7");
207 filter->Reset();
208 db->CompactRange(CompactRangeOptions(), &start, &end);
209 ASSERT_EQ(0, filter->NumKeys());
210 }
211
212 {
213 // L0: 1, 2, [4, 8]
214 // [3, 7] overlaps with 4 in L0
215 Slice start("3");
216 Slice end("7");
217 filter->Reset();
218 db->CompactRange(CompactRangeOptions(), &start, &end);
219 ASSERT_EQ(2, filter->NumKeys());
220 ASSERT_EQ(0, filter->KeyLevel("4"));
221 ASSERT_EQ(0, filter->KeyLevel("8"));
222 }
223
224 {
225 // L0: 1, 2
226 // L1: [4, 8]
227 // no file has keys in range (-inf, 0]
228 Slice end("0");
229 filter->Reset();
230 db->CompactRange(CompactRangeOptions(), nullptr, &end);
231 ASSERT_EQ(0, filter->NumKeys());
232 }
233
234 {
235 // L0: 1, 2
236 // L1: [4, 8]
237 // no file has keys in range [9, inf)
238 Slice start("9");
239 filter->Reset();
240 db->CompactRange(CompactRangeOptions(), &start, nullptr);
241 ASSERT_EQ(0, filter->NumKeys());
242 }
243
244 {
245 // L0: 1, 2
246 // L1: [4, 8]
247 // [2, 2] overlaps with 2 in L0
248 Slice start("2");
249 Slice end("2");
250 filter->Reset();
251 db->CompactRange(CompactRangeOptions(), &start, &end);
252 ASSERT_EQ(1, filter->NumKeys());
253 ASSERT_EQ(0, filter->KeyLevel("2"));
254 }
255
256 {
257 // L0: 1
258 // L1: 2, [4, 8]
259 // [2, 5] overlaps with 2 and [4, 8) in L1, skip L0
260 Slice start("2");
261 Slice end("5");
262 filter->Reset();
263 db->CompactRange(CompactRangeOptions(), &start, &end);
264 ASSERT_EQ(3, filter->NumKeys());
265 ASSERT_EQ(1, filter->KeyLevel("2"));
266 ASSERT_EQ(1, filter->KeyLevel("4"));
267 ASSERT_EQ(1, filter->KeyLevel("8"));
268 }
269
270 {
271 // L0: 1
272 // L1: [2, 4, 8]
273 // [0, inf) overlaps all files
274 Slice start("0");
275 filter->Reset();
276 db->CompactRange(CompactRangeOptions(), &start, nullptr);
277 ASSERT_EQ(4, filter->NumKeys());
278 // 1 is first compacted to L1 and then further compacted into [2, 4, 8],
279 // so finally the logged level for 1 is L1.
280 ASSERT_EQ(1, filter->KeyLevel("1"));
281 ASSERT_EQ(1, filter->KeyLevel("2"));
282 ASSERT_EQ(1, filter->KeyLevel("4"));
283 ASSERT_EQ(1, filter->KeyLevel("8"));
284 }
285
286 delete filter;
287 delete db;
288 DestroyDB(dbname_, options);
289 }
290
291 } // anonymous namespace
292
main(int argc,char ** argv)293 int main(int argc, char** argv) {
294 ::testing::InitGoogleTest(&argc, argv);
295 return RUN_ALL_TESTS();
296 }
297