Made merge_oprator a shared_ptr; and added TTL unit tests

Test Plan:
- make all check;
- make release;
- make stringappend_test; ./stringappend_test

Reviewers: haobo, emayanke

Reviewed By: haobo

CC: leveldb, kailiu

Differential Revision: https://reviews.facebook.net/D12381

db/builder.cc

@@ -54,7 +54,7 @@ Status BuildTable(const std::string& dbname,
     meta->smallest_seqno = GetInternalKeySeqno(key);
     meta->largest_seqno = meta->smallest_seqno;
 
-    MergeHelper merge(user_comparator, options.merge_operator,
+    MergeHelper merge(user_comparator, options.merge_operator.get(),
                       options.info_log.get(),
                       true /* internal key corruption is not ok */);
 

db/db_bench.cc

@@ -637,7 +637,6 @@ class Benchmark {
   int key_size_;
   int entries_per_batch_;
   WriteOptions write_options_;
-  std::shared_ptr<MergeOperator> merge_operator_;
   long reads_;
   long writes_;
   long readwrites_;
@@ -779,7 +778,6 @@ class Benchmark {
     value_size_(FLAGS_value_size),
     key_size_(FLAGS_key_size),
     entries_per_batch_(1),
-    merge_operator_(nullptr),
     reads_(FLAGS_reads < 0 ? FLAGS_num : FLAGS_reads),
     writes_(FLAGS_writes < 0 ? FLAGS_num : FLAGS_writes),
     readwrites_((FLAGS_writes < 0  && FLAGS_reads < 0)? FLAGS_num :
@@ -1212,13 +1210,13 @@ class Benchmark {
     options.bytes_per_sync = FLAGS_bytes_per_sync;
 
     // merge operator options
-    merge_operator_ = MergeOperators::CreateFromStringId(FLAGS_merge_operator);
+    options.merge_operator = MergeOperators::CreateFromStringId(
-    if (merge_operator_ == nullptr && !FLAGS_merge_operator.empty()) {
+        FLAGS_merge_operator);
+    if (options.merge_operator == nullptr && !FLAGS_merge_operator.empty()) {
       fprintf(stderr, "invalid merge operator: %s\n",
               FLAGS_merge_operator.c_str());
       exit(1);
     }
-    options.merge_operator = merge_operator_.get();
 
     Status s;
     if(FLAGS_read_only) {

db/db_impl.cc

@@ -1777,7 +1777,7 @@ Status DBImpl::DoCompactionWork(CompactionState* compact) {
   SequenceNumber visible_in_snapshot = kMaxSequenceNumber;
   std::string compaction_filter_value;
   std::vector<char> delete_key; // for compaction filter
-  MergeHelper merge(user_comparator(), options_.merge_operator,
+  MergeHelper merge(user_comparator(), options_.merge_operator.get(),
                     options_.info_log.get(),
                     false /* internal key corruption is expected */);
   auto compaction_filter = options_.compaction_filter;

db/db_iter.cc

@@ -57,7 +57,7 @@ class DBIter: public Iterator {
         env_(env),
         logger_(options.info_log),
         user_comparator_(cmp),
-        user_merge_operator_(options.merge_operator),
+        user_merge_operator_(options.merge_operator.get()),
         iter_(iter),
         sequence_(s),
         direction_(kForward),

db/db_test.cc

@@ -224,8 +224,6 @@ class DBTest {
   };
   int option_config_;
 
-  std::shared_ptr<MergeOperator> merge_operator_;
-
  public:
   std::string dbname_;
   SpecialEnv* env_;
@@ -242,7 +240,6 @@ class DBTest {
   };
 
   DBTest() : option_config_(kDefault),
-             merge_operator_(MergeOperators::CreatePutOperator()),
              env_(new SpecialEnv(Env::Default())) {
     filter_policy_ = NewBloomFilterPolicy(10);
     dbname_ = test::TmpDir() + "/db_test";
@@ -297,7 +294,7 @@ class DBTest {
     Options options;
     switch (option_config_) {
       case kMergePut:
-        options.merge_operator = merge_operator_.get();
+        options.merge_operator = MergeOperators::CreatePutOperator();
         break;
       case kFilter:
         options.filter_policy = filter_policy_;

db/memtable.cc

@@ -139,7 +139,7 @@ bool MemTable::Get(const LookupKey& key, std::string* value, Status* s,
   assert(operands != nullptr);
 
   bool merge_in_progress = s->IsMergeInProgress();
-  auto merge_operator = options.merge_operator;
+  auto merge_operator = options.merge_operator.get();
   auto logger = options.info_log;
   std::string merge_result;
 

db/merge_test.cc

@@ -16,14 +16,13 @@
 using namespace std;
 using namespace leveldb;
 
-auto mergeOperator = MergeOperators::CreateUInt64AddOperator();
 
 std::shared_ptr<DB> OpenDb(const string& dbname, const bool ttl = false) {
   DB* db;
   StackableDB* sdb;
   Options options;
   options.create_if_missing = true;
-  options.merge_operator = mergeOperator.get();
+  options.merge_operator = MergeOperators::CreateUInt64AddOperator();
   Status s;
   DestroyDB(dbname, Options());
   if (ttl) {

db/version_set.cc

@@ -381,7 +381,7 @@ void Version::Get(const ReadOptions& options,
   Slice user_key = k.user_key();
   const Comparator* ucmp = vset_->icmp_.user_comparator();
 
-  auto merge_operator = db_options.merge_operator;
+  auto merge_operator = db_options.merge_operator.get();
   auto logger = db_options.info_log;
 
   assert(status->ok() || status->IsMergeInProgress());

include/leveldb/options.h

@@ -85,7 +85,7 @@ struct Options {
   // for the first time. It's necessary to specify a merge operator when
   // openning the DB in this case.
   // Default: nullptr
-  const MergeOperator* merge_operator;
+  shared_ptr<MergeOperator> merge_operator;
 
   // The client must provide compaction_filter_factory if it requires a new
   // compaction filter to be used for different compaction processes

tools/db_stress.cc

@@ -537,7 +537,6 @@ class StressTest {
                           FLAGS_test_batches_snapshots ?
                           sizeof(long) : sizeof(long)-1)),
         db_(nullptr),
-        merge_operator_(MergeOperators::CreatePutOperator()),
         num_times_reopened_(0) {
     if (FLAGS_destroy_db_initially) {
       std::vector<std::string> files;
@@ -553,7 +552,6 @@ class StressTest {
 
   ~StressTest() {
     delete db_;
-    merge_operator_ = nullptr;
     delete filter_policy_;
     delete prefix_extractor_;
   }
@@ -1140,7 +1138,7 @@ class StressTest {
     }
 
     if (FLAGS_use_merge_put) {
-      options.merge_operator = merge_operator_.get();
+      options.merge_operator = MergeOperators::CreatePutOperator();
     }
 
     fprintf(stdout, "DB path: [%s]\n", FLAGS_db);
@@ -1188,7 +1186,6 @@ class StressTest {
   const SliceTransform* prefix_extractor_;
   DB* db_;
   StackableDB* sdb_;
-  std::shared_ptr<MergeOperator> merge_operator_;
   int num_times_reopened_;
 };
 

utilities/merge_operators/string_append/stringappend_test.cc

@@ -13,6 +13,8 @@
 #include "leveldb/merge_operator.h"
 #include "utilities/merge_operators.h"
 #include "utilities/merge_operators/string_append/stringappend.h"
+#include "utilities/merge_operators/string_append/stringappend2.h"
+#include "utilities/ttl/db_ttl.h"
 #include "util/testharness.h"
 #include "util/random.h"
 
@@ -24,16 +26,24 @@ namespace leveldb {
 const std::string kDbName = "/tmp/mergetestdb";
 
 // OpenDb opens a (possibly new) rocksdb database with a StringAppendOperator
-std::shared_ptr<DB> OpenDb(StringAppendOperator* append_op) {
+std::shared_ptr<DB> OpenNormalDb(char delim_char) {
   DB* db;
   Options options;
   options.create_if_missing = true;
-    options.merge_operator = append_op;
+  options.merge_operator.reset(new StringAppendOperator(delim_char));
-    Status s = DB::Open(options, kDbName,  &db);
+  ASSERT_OK(DB::Open(options, kDbName,  &db));
-    if (!s.ok()) {
+  return std::shared_ptr<DB>(db);
-      std::cerr << s.ToString() << std::endl;
-      assert(false);
 }
+
+// Open a TtlDB with a non-associative StringAppendTESTOperator
+std::shared_ptr<DB> OpenTtlDb(char delim_char) {
+  StackableDB* db;
+  Options options;
+  options.create_if_missing = true;
+  options.merge_operator.reset(new StringAppendTESTOperator(delim_char));
+  Status s;
+  db = new DBWithTTL(123456, options, kDbName, s, false);
+  ASSERT_OK(s);
   return std::shared_ptr<DB>(db);
 }
 
@@ -66,7 +76,6 @@ class StringLists {
   // Returns the list of strings associated with key (or "" if does not exist)
   bool Get(const std::string& key, std::string* const result){
     assert(result != NULL); // we should have a place to store the result
-
     auto s = db_->Get(get_option_, key, result);
 
     if (s.ok()) {
@@ -86,6 +95,7 @@ class StringLists {
     return false;
   }
 
+
  private:
   std::shared_ptr<DB> db_;
   WriteOptions merge_option_;
@@ -93,14 +103,31 @@ class StringLists {
 
 };
 
+
+// The class for unit-testing
+class StringAppendOperatorTest {
+ public:
+  StringAppendOperatorTest() {
+    DestroyDB(kDbName, Options());    // Start each test with a fresh DB
+  }
+
+  typedef std::shared_ptr<DB> (* OpenFuncPtr)(char);
+
+  // Allows user to open databases with different configurations.
+  // e.g.: Can open a DB or a TtlDB, etc.
+  static void SetOpenDbFunction(OpenFuncPtr func) {
+    OpenDb = func;
+  }
+
+ protected:
+  static OpenFuncPtr OpenDb;
+};
+StringAppendOperatorTest::OpenFuncPtr StringAppendOperatorTest::OpenDb = nullptr;
+
 // THE TEST CASES BEGIN HERE
 
-class StringAppendOperatorTest { };
 TEST(StringAppendOperatorTest, IteratorTest) {
-  DestroyDB(kDbName, Options());    // Start this test with a fresh DB
+  auto db_ = OpenDb(',');
-
-  StringAppendOperator append_op(',');
-  auto db_ = OpenDb(&append_op);
   StringLists slists(db_);
 
   slists.Append("k1", "v1");
@@ -193,10 +220,7 @@ TEST(StringAppendOperatorTest, IteratorTest) {
 }
 
 TEST(StringAppendOperatorTest, SimpleTest) {
-  DestroyDB(kDbName, Options());    // Start this test with a fresh DB
+  auto db = OpenDb(',');
-
-  StringAppendOperator append_op(',');
-  auto db = OpenDb(&append_op);
   StringLists slists(db);
 
   slists.Append("k1", "v1");
@@ -211,10 +235,7 @@ TEST(StringAppendOperatorTest,SimpleTest) {
 }
 
 TEST(StringAppendOperatorTest, SimpleDelimiterTest) {
-  DestroyDB(kDbName, Options());    // Start this test with a fresh DB
+  auto db = OpenDb('|');
-
-  StringAppendOperator append_op('|');
-  auto db = OpenDb(&append_op);
   StringLists slists(db);
 
   slists.Append("k1", "v1");
@@ -227,10 +248,7 @@ TEST(StringAppendOperatorTest,SimpleDelimiterTest) {
 }
 
 TEST(StringAppendOperatorTest, OneValueNoDelimiterTest) {
-  DestroyDB(kDbName, Options());    // Start this test with a fresh DB
+  auto db = OpenDb('!');
-
-  StringAppendOperator append_op('!');
-  auto db = OpenDb(&append_op);
   StringLists slists(db);
 
   slists.Append("random_key", "single_val");
@@ -241,10 +259,7 @@ TEST(StringAppendOperatorTest,OneValueNoDelimiterTest) {
 }
 
 TEST(StringAppendOperatorTest, VariousKeys) {
-  DestroyDB(kDbName, Options());    // Start this test with a fresh DB
+  auto db = OpenDb('\n');
-
-  StringAppendOperator append_op('\n');
-  auto db = OpenDb(&append_op);
   StringLists slists(db);
 
   slists.Append("c", "asdasd");
@@ -270,10 +285,7 @@ TEST(StringAppendOperatorTest,VariousKeys) {
 
 // Generate semi random keys/words from a small distribution.
 TEST(StringAppendOperatorTest, RandomMixGetAppend) {
-  DestroyDB(kDbName, Options());    // Start this test with a fresh DB
+  auto db = OpenDb(' ');
-
-  StringAppendOperator append_op(' ');
-  auto db = OpenDb(&append_op);
   StringLists slists(db);
 
   // Generate a list of random keys and values
@@ -299,9 +311,6 @@ TEST(StringAppendOperatorTest,RandomMixGetAppend) {
     std::string key = keys[randomGen.Uniform((int)kKeyCount)];
     std::string word = words[randomGen.Uniform((int)kWordCount)];
 
-    // Debug message.
-    //std::cout << (int)query << " " << key << " " << word << std::endl;
-
     // Apply the query and any checks.
     if (query == APPEND_OP) {
 
@@ -327,10 +336,7 @@ TEST(StringAppendOperatorTest,RandomMixGetAppend) {
 }
 
 TEST(StringAppendOperatorTest, BIGRandomMixGetAppend) {
-  DestroyDB(kDbName, Options());    // Start this test with a fresh DB
+  auto db = OpenDb(' ');
-
-  StringAppendOperator append_op(' ');
-  auto db = OpenDb(&append_op);
   StringLists slists(db);
 
   // Generate a list of random keys and values
@@ -356,9 +362,6 @@ TEST(StringAppendOperatorTest,BIGRandomMixGetAppend) {
     std::string key = keys[randomGen.Uniform((int)kKeyCount)];
     std::string word = words[randomGen.Uniform((int)kWordCount)];
 
-    // Debug message.
-    //std::cout << (int)query << " " << key << " " << word << std::endl;
-
     //Apply the query and any checks.
     if (query == APPEND_OP) {
 
@@ -385,12 +388,9 @@ TEST(StringAppendOperatorTest,BIGRandomMixGetAppend) {
 
 
 TEST(StringAppendOperatorTest, PersistentVariousKeys) {
-  DestroyDB(kDbName, Options());    // Start this test with a fresh DB
-
   // Perform the following operations in limited scope
   {
-    StringAppendOperator append_op('\n');
+    auto db = OpenDb('\n');
-    auto db = OpenDb(&append_op);
     StringLists slists(db);
 
     slists.Append("c", "asdasd");
@@ -413,8 +413,7 @@ TEST(StringAppendOperatorTest,PersistentVariousKeys) {
 
   // Reopen the database (the previous changes should persist / be remembered)
   {
-    StringAppendOperator append_op('\n');
+    auto db = OpenDb('\n');
-    auto db = OpenDb(&append_op);
     StringLists slists(db);
 
     slists.Append("c", "bbnagnagsx");
@@ -440,8 +439,7 @@ TEST(StringAppendOperatorTest,PersistentVariousKeys) {
 
   // Reopen the database (the previous changes should persist / be remembered)
   {
-    StringAppendOperator append_op('\n');
+    auto db = OpenDb('\n');
-    auto db = OpenDb(&append_op);
     StringLists slists(db);
 
     // All changes should be on disk. This will test VersionSet Get()
@@ -457,13 +455,9 @@ TEST(StringAppendOperatorTest,PersistentVariousKeys) {
 }
 
 TEST(StringAppendOperatorTest, PersistentFlushAndCompaction) {
-  DestroyDB(kDbName, Options());    // Start this test with a fresh DB
-
-  StringAppendOperator append_op('\n');
-
   // Perform the following operations in limited scope
   {
-    auto db = OpenDb(&append_op);
+    auto db = OpenDb('\n');
     StringLists slists(db);
     std::string a, b, c;
     bool success;
@@ -509,7 +503,7 @@ TEST(StringAppendOperatorTest,PersistentFlushAndCompaction) {
 
   // Reopen the database (the previous changes should persist / be remembered)
   {
-    auto db = OpenDb(&append_op);
+    auto db = OpenDb('\n');
     StringLists slists(db);
     std::string a, b, c;
 
@@ -557,10 +551,7 @@ TEST(StringAppendOperatorTest,PersistentFlushAndCompaction) {
 }
 
 TEST(StringAppendOperatorTest, SimpleTestNullDelimiter) {
-  DestroyDB(kDbName, Options());    // Start this test with a fresh DB
+  auto db = OpenDb('\0');
-
-  StringAppendOperator append_op('\0');
-  auto db = OpenDb(&append_op);
   StringLists slists(db);
 
   slists.Append("k1", "v1");
@@ -582,10 +573,22 @@ TEST(StringAppendOperatorTest,SimpleTestNullDelimiter) {
   ASSERT_EQ(res, checker);
 }
 
-
 } // namespace leveldb
 
 int main(int arc, char** argv) {
+  // Run with regular database
+  {
+    fprintf(stderr, "Running tests with regular db and operator.\n");
+    StringAppendOperatorTest::SetOpenDbFunction(&OpenNormalDb);
     leveldb::test::RunAllTests();
+  }
+
+  // Run with TTL
+  {
+    fprintf(stderr, "Running tests with ttl db and generic operator.\n");
+    StringAppendOperatorTest::SetOpenDbFunction(&OpenTtlDb);
+    leveldb::test::RunAllTests();
+  }
+
   return 0;
 }

utilities/ttl/db_ttl.cc

@@ -32,8 +32,8 @@ DBWithTTL::DBWithTTL(const int32_t ttl,
   }
 
   if (options.merge_operator) {
-    ttl_merge_op_.reset(new TtlMergeOperator(options.merge_operator));
+    options_to_open.merge_operator.reset(
-    options_to_open.merge_operator = ttl_merge_op_.get();
+      new TtlMergeOperator(options.merge_operator));
   }
 
   if (read_only) {

utilities/ttl/db_ttl.h

@@ -112,7 +112,6 @@ class DBWithTTL : public StackableDB {
  private:
   DB* db_;
   int32_t ttl_;
-  unique_ptr<MergeOperator> ttl_merge_op_;
   unique_ptr<CompactionFilter> ttl_comp_filter_;
 };
 
@@ -258,7 +257,7 @@ class TtlCompactionFilterFactory : public CompactionFilterFactory {
 class TtlMergeOperator : public MergeOperator {
 
  public:
-  explicit TtlMergeOperator(const MergeOperator* merge_op)
+  explicit TtlMergeOperator(const std::shared_ptr<MergeOperator> merge_op)
     : user_merge_op_(merge_op) {
     assert(merge_op);
   }
@@ -356,7 +355,7 @@ class TtlMergeOperator : public MergeOperator {
   }
 
  private:
-  const MergeOperator* user_merge_op_;
+  std::shared_ptr<MergeOperator> user_merge_op_;
 };
 
 }