ceph/src/rocksdb/db/manual_compaction_test.cc

   1 //  Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
   2 //  This source code is licensed under the BSD-style license found in the
   3 //  LICENSE file in the root directory of this source tree. An additional grant
   4 //  of patent rights can be found in the PATENTS file in the same directory.
   5 //
   6 // Test for issue 178: a manual compaction causes deleted data to reappear.
   7 #include <iostream>
   8 #include <sstream>
   9 #include <cstdlib>
  10
  11 #include "rocksdb/db.h"
  12 #include "rocksdb/compaction_filter.h"
  13 #include "rocksdb/slice.h"
  14 #include "rocksdb/write_batch.h"
  15 #include "util/testharness.h"
  16 #include "port/port.h"
  17
  18 using namespace rocksdb;
  19
  20 namespace {
  21
  22 const int kNumKeys = 1100000;
  23
  24 std::string Key1(int i) {
  25   char buf[100];
  26   snprintf(buf, sizeof(buf), "my_key_%d", i);
  27   return buf;
  28 }
  29
  30 std::string Key2(int i) {
  31   return Key1(i) + "_xxx";
  32 }
  33
  34 class ManualCompactionTest : public testing::Test {
  35  public:
  36   ManualCompactionTest() {
  37     // Get rid of any state from an old run.
  38     dbname_ = rocksdb::test::TmpDir() + "/rocksdb_cbug_test";
  39     DestroyDB(dbname_, rocksdb::Options());
  40   }
  41
  42   std::string dbname_;
  43 };
  44
  45 class DestroyAllCompactionFilter : public CompactionFilter {
  46  public:
  47   DestroyAllCompactionFilter() {}
  48
  49   virtual bool Filter(int level, const Slice& key, const Slice& existing_value,
  50                       std::string* new_value,
  51                       bool* value_changed) const override {
  52     return existing_value.ToString() == "destroy";
  53   }
  54
  55   virtual const char* Name() const override {
  56     return "DestroyAllCompactionFilter";
  57   }
  58 };
  59
  60 TEST_F(ManualCompactionTest, CompactTouchesAllKeys) {
  61   for (int iter = 0; iter < 2; ++iter) {
  62     DB* db;
  63     Options options;
  64     if (iter == 0) { // level compaction
  65       options.num_levels = 3;
  66       options.compaction_style = kCompactionStyleLevel;
  67     } else { // universal compaction
  68       options.compaction_style = kCompactionStyleUniversal;
  69     }
  70     options.create_if_missing = true;
  71     options.compression = rocksdb::kNoCompression;
  72     options.compaction_filter = new DestroyAllCompactionFilter();
  73     ASSERT_OK(DB::Open(options, dbname_, &db));
  74
  75     db->Put(WriteOptions(), Slice("key1"), Slice("destroy"));
  76     db->Put(WriteOptions(), Slice("key2"), Slice("destroy"));
  77     db->Put(WriteOptions(), Slice("key3"), Slice("value3"));
  78     db->Put(WriteOptions(), Slice("key4"), Slice("destroy"));
  79
  80     Slice key4("key4");
  81     db->CompactRange(CompactRangeOptions(), nullptr, &key4);
  82     Iterator* itr = db->NewIterator(ReadOptions());
  83     itr->SeekToFirst();
  84     ASSERT_TRUE(itr->Valid());
  85     ASSERT_EQ("key3", itr->key().ToString());
  86     itr->Next();
  87     ASSERT_TRUE(!itr->Valid());
  88     delete itr;
  89
  90     delete options.compaction_filter;
  91     delete db;
  92     DestroyDB(dbname_, options);
  93   }
  94 }
  95
  96 TEST_F(ManualCompactionTest, Test) {
  97   // Open database.  Disable compression since it affects the creation
  98   // of layers and the code below is trying to test against a very
  99   // specific scenario.
 100   rocksdb::DB* db;
 101   rocksdb::Options db_options;
 102   db_options.create_if_missing = true;
 103   db_options.compression = rocksdb::kNoCompression;
 104   ASSERT_OK(rocksdb::DB::Open(db_options, dbname_, &db));
 105
 106   // create first key range
 107   rocksdb::WriteBatch batch;
 108   for (int i = 0; i < kNumKeys; i++) {
 109     batch.Put(Key1(i), "value for range 1 key");
 110   }
 111   ASSERT_OK(db->Write(rocksdb::WriteOptions(), &batch));
 112
 113   // create second key range
 114   batch.Clear();
 115   for (int i = 0; i < kNumKeys; i++) {
 116     batch.Put(Key2(i), "value for range 2 key");
 117   }
 118   ASSERT_OK(db->Write(rocksdb::WriteOptions(), &batch));
 119
 120   // delete second key range
 121   batch.Clear();
 122   for (int i = 0; i < kNumKeys; i++) {
 123     batch.Delete(Key2(i));
 124   }
 125   ASSERT_OK(db->Write(rocksdb::WriteOptions(), &batch));
 126
 127   // compact database
 128   std::string start_key = Key1(0);
 129   std::string end_key = Key1(kNumKeys - 1);
 130   rocksdb::Slice least(start_key.data(), start_key.size());
 131   rocksdb::Slice greatest(end_key.data(), end_key.size());
 132
 133   // commenting out the line below causes the example to work correctly
 134   db->CompactRange(CompactRangeOptions(), &least, &greatest);
 135
 136   // count the keys
 137   rocksdb::Iterator* iter = db->NewIterator(rocksdb::ReadOptions());
 138   int num_keys = 0;
 139   for (iter->SeekToFirst(); iter->Valid(); iter->Next()) {
 140     num_keys++;
 141   }
 142   delete iter;
 143   ASSERT_EQ(kNumKeys, num_keys) << "Bad number of keys";
 144
 145   // close database
 146   delete db;
 147   DestroyDB(dbname_, rocksdb::Options());
 148 }
 149
 150 }  // anonymous namespace
 151
 152 int main(int argc, char** argv) {
 153   ::testing::InitGoogleTest(&argc, argv);
 154   return RUN_ALL_TESTS();
 155 }