Adding Sanity checks in Get and Iterator->value for ttl

Summary: The 2 checks added will increase reliabilty and help in debugging

Test Plan: make ttl_test;./ttl_test

Reviewers: vamsi, dhruba, sheki, haobo

Reviewed By: vamsi

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

Makefile

@@ -53,6 +53,7 @@ TESTS = \
 	memenv_test \
 	skiplist_test \
 	table_test \
+	ttl_test \
 	block_test \
 	version_edit_test \
 	version_set_test \
@@ -182,6 +183,9 @@ crc32c_test: util/crc32c_test.o $(LIBOBJECTS) $(TESTHARNESS)
 db_test: db/db_test.o $(LIBOBJECTS) $(TESTHARNESS)
 	$(CXX) db/db_test.o $(LIBOBJECTS) $(TESTHARNESS) $(EXEC_LDFLAGS) -o $@ $(LDFLAGS)
 
+ttl_test: utilities/ttl/ttl_test.o $(LIBOBJECTS) $(TESTHARNESS)
+	$(CXX) utilities/ttl/ttl_test.o $(LIBOBJECTS) $(TESTHARNESS) $(EXEC_LDFLAGS) -o $@  $(LDFLAGS)
+
 dbformat_test: db/dbformat_test.o $(LIBOBJECTS) $(TESTHARNESS)
 	$(CXX) db/dbformat_test.o $(LIBOBJECTS) $(TESTHARNESS) $(EXEC_LDFLAGS) -o $@ $(LDFLAGS)
 

build_detect_platform

@@ -130,7 +130,7 @@ esac
 # except for the test and benchmark files. By default, find will output a list
 # of all files matching either rule, so we need to append -print to make the
 # prune take effect.
-DIRS="util db table"
+DIRS="util db table utilities"
 if test "$USE_THRIFT"; then
   DIRS="$DIRS thrift/server_utils.cpp thrift/gen-cpp "
   THRIFTSERVER=leveldb_server

db/db_impl.cc

@@ -1579,6 +1579,7 @@ Status DBImpl::DoCompactionWork(CompactionState* compact) {
   SequenceNumber last_sequence_for_key __attribute__((unused)) =
     kMaxSequenceNumber;
   SequenceNumber visible_in_snapshot = kMaxSequenceNumber;
+  std::string compaction_filter_value;
   for (; input->Valid() && !shutting_down_.Acquire_Load(); ) {
     // Prioritize immutable compaction work
     if (imm_.imm_flush_needed.NoBarrier_Load() != nullptr) {
@@ -1594,7 +1595,7 @@ Status DBImpl::DoCompactionWork(CompactionState* compact) {
 
     Slice key = input->key();
     Slice value = input->value();
-    Slice* compaction_filter_value = nullptr;
+
     if (compact->compaction->ShouldStopBefore(key) &&
         compact->builder != nullptr) {
       status = FinishCompactionOutputFile(compact, input.get());
@@ -1653,20 +1654,24 @@ Status DBImpl::DoCompactionWork(CompactionState* compact) {
                  ikey.type != kTypeDeletion &&
                  ikey.sequence < earliest_snapshot) {
         // If the user has specified a compaction filter, then invoke
-        // it. If this key is not visible via any snapshot and the
-        // return value of the compaction filter is true and then
+        // it. If the return value of the compaction filter is true,
         // drop this key from the output.
+        bool value_changed = false;
+        compaction_filter_value.clear();
         drop = options_.CompactionFilter(options_.compaction_filter_args,
-                         compact->compaction->level(),
-                         ikey.user_key, value, &compaction_filter_value);
-
+                                         compact->compaction->level(),
+                                         ikey.user_key, value,
+                                         &compaction_filter_value,
+                                         &value_changed);
+        // Another example of statistics update without holding the lock
+        // TODO: clean it up
         if (drop) {
           RecordTick(options_.statistics, COMPACTION_KEY_DROP_USER);
         }
+
         // If the application wants to change the value, then do so here.
-        if (compaction_filter_value != nullptr) {
-          value = *compaction_filter_value;
-          delete compaction_filter_value;
+        if (value_changed) {
+          value = compaction_filter_value;
         }
       }
 

db/db_test.cc

@@ -1351,22 +1351,26 @@ TEST(DBTest, RepeatedWritesToSameKey) {
 static int cfilter_count;
 static std::string NEW_VALUE = "NewValue";
 static bool keep_filter(void* arg, int level, const Slice& key,
-  const Slice& value, Slice** new_value) {
+                        const Slice& value, std::string* new_value,
+                        bool* value_changed) {
   assert(arg == nullptr);
   cfilter_count++;
   return false;
 }
 static bool delete_filter(void*argv, int level, const Slice& key,
-  const Slice& value, Slice** new_value) {
+                          const Slice& value, std::string* new_value,
+                          bool* value_changed) {
   assert(argv == nullptr);
   cfilter_count++;
   return true;
 }
 static bool change_filter(void*argv, int level, const Slice& key,
-  const Slice& value, Slice** new_value) {
+                          const Slice& value, std::string* new_value,
+                          bool* value_changed) {
   assert(argv == (void*)100);
   assert(new_value != nullptr);
-  *new_value = new Slice(NEW_VALUE);
+  *new_value = NEW_VALUE;
+  *value_changed = true;
   return false;
 }
 

include/leveldb/options.h

@@ -356,22 +356,25 @@ struct Options {
 
   // This method allows an application to modify/delete a key-value at
   // the time of compaction. The compaction process invokes this
-  // method for every kv that is being compacted. A return value
+  // method for kv that is being compacted. A return value
   // of false indicates that the kv should be preserved in the
   // output of this compaction run and a return value of true
   // indicates that this key-value should be removed from the
   // output of the compaction.  The application can inspect
-  // the existing value of the key, modify it if needed and
-  // return back the new value for this key. The application
-  // should allocate memory for the Slice object that is used to
-  // return the new value and the leveldb framework will
-  // free up that memory.
+  // the existing value of the key and make decision based on it.
+
+  // When the value is to be preserved, the application has the option
+  // to modify the existing_value and pass it back through new_value.
+  // value_changed needs to be set to true in this case.
+
   // The compaction_filter_args, if specified here, are passed
   // back to the invocation of the CompactionFilter.
   void* compaction_filter_args;
   bool (*CompactionFilter)(void* compaction_filter_args,
-         int level, const Slice& key,
-         const Slice& existing_value, Slice** new_value);
+                           int level, const Slice& key,
+                           const Slice& existing_value,
+                           std::string* new_value,
+                           bool* value_changed);
 
   // Disable automatic compactions. Manual compactions can still
   // be issued on this database.

include/utilities/utility_db.h

@@ -0,0 +1,52 @@
+// Copyright (c) 2011 The LevelDB Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file. See the AUTHORS file for names of contributors.
+
+#ifndef LEVELDB_INCLUDE_UTILITIES_UTILITY_DB_H_
+#define LEVELDB_INCLUDE_UTILITIES_UTILITY_DB_H_
+
+#include "include/leveldb/db.h"
+
+namespace leveldb {
+
+// This class contains APIs to open leveldb with specific support eg. TTL
+class UtilityDB {
+
+  public:
+    // Open the database with TTL support.
+    //
+    // USE-CASES:
+    // This API should be used to open the db when key-values inserted are
+    //  meant to be removed from the db in a non-strict 'ttl' amount of time
+    //  Therefore, this guarantees that key-values inserted will remain in the
+    //  db for >= ttl amount of time and the db will make efforts to remove the
+    //  key-values as soon as possible after ttl seconds of their insertion.
+    //
+    // BEHAVIOUR:
+    // TTL is accepted in seconds
+    // (int32_t)Timestamp(creation) is suffixed to values in Put internally
+    // Expired TTL values deleted in compaction only:(Timestamp+ttl<time_now)
+    // Get/Iterator may return expired entries(compaction not run on them yet)
+    // Different TTL may be used during different Opens
+    // Example: Open1 at t=0 with ttl=4 and insert k1,k2, close at t=2
+    //          Open2 at t=3 with ttl=5. Now k1,k2 should be deleted at t>=5
+    //
+    // CONSTRAINTS:
+    // The caller must not specify any compaction-filter in options
+    // Not specifying/passing or non-positive TTL behaves like TTL = infinity
+    //
+    // !!!WARNING!!!:
+    // Calling DB::Open directly to re-open a db created by this API will get
+    //  corrupt values(timestamp suffixed) and no ttl effect will be there
+    //  during the second Open, so use this API consistently to open the db
+    // Be careful when passing ttl with a small positive value because the
+    //  whole database may be deleted in a small amount of time
+    static Status OpenTtlDB(const Options& options,
+                            const std::string& name,
+                            DB** dbptr,
+                            int32_t ttl = 0);
+};
+
+} //  namespace leveldb
+
+#endif  //  LEVELDB_INCLUDE_UTILITIES_UTILITY_DB_H_

tools/db_stress.cc

@@ -21,7 +21,7 @@
 #include "db/version_set.h"
 #include "db/db_statistics.h"
 #include "leveldb/cache.h"
-#include "leveldb/db.h"
+#include "utilities/utility_db.h"
 #include "leveldb/env.h"
 #include "leveldb/write_batch.h"
 #include "leveldb/statistics.h"
@@ -31,6 +31,8 @@
 #include "util/mutexlock.h"
 #include "util/random.h"
 #include "util/testutil.h"
+#include "util/logging.h"
+#include "utilities/ttl/db_ttl.h"
 #include "hdfs/env_hdfs.h"
 
 static const long KB = 1024;
@@ -54,6 +56,11 @@ static bool FLAGS_test_batches_snapshots = false;
 // Number of concurrent threads to run.
 static int FLAGS_threads = 32;
 
+// Opens the db with this ttl value if this is not -1
+// Carefully specify a large value such that verifications on deleted
+//  values don't fail
+static int FLAGS_ttl = -1;
+
 // Size of each value will be this number times rand_int(1,3) bytes
 static int FLAGS_value_size_mult = 8;
 
@@ -870,6 +877,11 @@ class StressTest {
             kMajorVersion, kMinorVersion);
     fprintf(stdout, "Number of threads   : %d\n", FLAGS_threads);
     fprintf(stdout, "Ops per thread      : %d\n", FLAGS_ops_per_thread);
+    std::string ttl_state("unused");
+    if (FLAGS_ttl > 0) {
+      ttl_state = NumberToString(FLAGS_ttl);
+    }
+    fprintf(stdout, "Time to live(sec)   : %s\n", ttl_state.c_str());
     fprintf(stdout, "Read percentage     : %d\n", FLAGS_readpercent);
     fprintf(stdout, "Write-buffer-size   : %d\n", FLAGS_write_buffer_size);
     fprintf(stdout, "Delete percentage   : %d\n", FLAGS_delpercent);
@@ -936,7 +948,12 @@ class StressTest {
     if (purge_percent.Uniform(100) < FLAGS_purge_redundant_percent - 1) {
       options.purge_redundant_kvs_while_flush = false;
     }
-    Status s = DB::Open(options, FLAGS_db, &db_);
+    Status s;
+    if (FLAGS_ttl == -1) {
+      s = DB::Open(options, FLAGS_db, &db_);
+    } else {
+      s = UtilityDB::OpenTtlDB(options, FLAGS_db, &db_, FLAGS_ttl);
+    }
     if (!s.ok()) {
       fprintf(stderr, "open error: %s\n", s.ToString().c_str());
       exit(1);
@@ -946,7 +963,11 @@ class StressTest {
   void Reopen() {
     // do not close the db. Just delete the lock file. This
     // simulates a crash-recovery kind of situation.
-    ((DBImpl*) db_)->TEST_Destroy_DBImpl();
+    if (FLAGS_ttl != -1) {
+      ((DBWithTTL*) db_)->TEST_Destroy_DBWithTtl();
+    } else {
+      ((DBImpl*) db_)->TEST_Destroy_DBImpl();
+    }
     db_ = nullptr;
 
     num_times_reopened_++;
@@ -1017,6 +1038,8 @@ int main(int argc, char** argv) {
       FLAGS_test_batches_snapshots = n;
     } else if (sscanf(argv[i], "--threads=%d%c", &n, &junk) == 1) {
       FLAGS_threads = n;
+    } else if (sscanf(argv[i], "--ttl=%d%c", &n, &junk) == 1) {
+      FLAGS_ttl = n;
     } else if (sscanf(argv[i], "--value_size_mult=%d%c", &n, &junk) == 1) {
       FLAGS_value_size_mult = n;
     } else if (sscanf(argv[i], "--write_buffer_size=%d%c", &n, &junk) == 1) {

utilities/ttl/db_ttl.cc

@@ -0,0 +1,275 @@
+// Copyright (c) 2011 The LevelDB Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file. See the AUTHORS file for names of contributors.
+
+#include "utilities/ttl/db_ttl.h"
+#include "include/utilities/utility_db.h"
+#include "db/filename.h"
+#include "util/coding.h"
+#include "include/leveldb/env.h"
+#include "include/leveldb/iterator.h"
+
+namespace leveldb {
+
+class TtlIterator : public Iterator {
+
+ public:
+  TtlIterator(Iterator* iter, int32_t ts_len)
+    : iter_(iter),
+      ts_len_(ts_len) {
+    assert(iter_);
+  }
+
+  ~TtlIterator() {
+    delete iter_;
+  }
+
+  bool Valid() const {
+    return iter_->Valid();
+  }
+
+  void SeekToFirst() {
+    iter_->SeekToFirst();
+  }
+
+  void SeekToLast() {
+    iter_->SeekToLast();
+  }
+
+  void Seek(const Slice& target) {
+    iter_->Seek(target);
+  }
+
+  void Next() {
+    iter_->Next();
+  }
+
+  void Prev() {
+    iter_->Prev();
+  }
+
+  Slice key() const {
+    return iter_->key();
+  }
+
+  Slice value() const {
+    assert(DBWithTTL::SanityCheckTimestamp(iter_->value().ToString()).ok());
+    Slice trimmed_value = iter_->value();
+    trimmed_value.size_ -= ts_len_;
+    return trimmed_value;
+  }
+
+  Status status() const {
+    return iter_->status();
+  }
+
+ private:
+  Iterator* iter_;
+  int32_t ts_len_;
+};
+
+// Open the db inside DBWithTTL because options needs pointer to its ttl
+DBWithTTL::DBWithTTL(const int32_t ttl,
+                     const Options& options,
+                     const std::string& dbname,
+                     Status& st)
+    : ttl_(ttl) {
+  assert(options.CompactionFilter == nullptr);
+  Options options_to_open = options;
+  options_to_open.compaction_filter_args = &ttl_;
+  options_to_open.CompactionFilter = DeleteByTS;
+  st = DB::Open(options_to_open, dbname, &db_);
+}
+
+DBWithTTL::~DBWithTTL() {
+  delete db_;
+}
+
+Status UtilityDB::OpenTtlDB(
+    const Options& options,
+    const std::string& dbname,
+    DB** dbptr,
+    int32_t ttl) {
+  Status st;
+  *dbptr = new DBWithTTL(ttl, options, dbname, st);
+  if (!st.ok()) {
+    delete dbptr;
+  }
+  return st;
+}
+
+// returns true(i.e. key-value to be deleted) if its TS has expired based on ttl
+bool DBWithTTL::DeleteByTS(
+    void* args,
+    int level,
+    const Slice& key,
+    const Slice& old_val,
+    std::string* new_val,
+    bool* value_changed) {
+  return IsStale(old_val, *(int32_t*)args);
+}
+
+// Gives back the current time
+Status DBWithTTL::GetCurrentTime(int32_t& curtime) {
+  return Env::Default()->GetCurrentTime((int64_t*)&curtime);
+}
+
+// Appends the current timestamp to the string.
+// Returns false if could not get the current_time, true if append succeeds
+Status DBWithTTL::AppendTS(const Slice& val, std::string& val_with_ts) {
+  val_with_ts.reserve(kTSLength + val.size());
+  char ts_string[kTSLength];
+  int32_t curtime;
+  Status st = GetCurrentTime(curtime);
+  if (!st.ok()) {
+    return st;
+  }
+  EncodeFixed32(ts_string, curtime);
+  val_with_ts.append(val.data(), val.size());
+  val_with_ts.append(ts_string, kTSLength);
+  return st;
+}
+
+// Returns corruption if the length of the string is lesser than timestamp, or
+// timestamp refers to a time lesser than ttl-feature release time
+Status DBWithTTL::SanityCheckTimestamp(const std::string& str) {
+  if (str.length() < (unsigned)kTSLength) {
+    return Status::Corruption("Error: value's length less than timestamp's\n");
+  }
+  // Checks that TS is not lesser than kMinTimestamp
+  // Gaurds against corruption & normal database opened incorrectly in ttl mode
+  int32_t timestamp_value =
+    DecodeFixed32(str.data() + str.size() - kTSLength);
+  if (timestamp_value < kMinTimestamp){
+    return Status::Corruption("Error: Timestamp < ttl feature release time!\n");
+  }
+  return Status::OK();
+}
+
+// Checks if the string is stale or not according to TTl provided
+bool DBWithTTL::IsStale(const Slice& value, int32_t ttl) {
+  if (ttl <= 0) { // Data is fresh if TTL is non-positive
+    return false;
+  }
+  int32_t curtime;
+  if (!GetCurrentTime(curtime).ok()) {
+    return false; // Treat the data as fresh if could not get current time
+  } else {
+    int32_t timestamp_value =
+      DecodeFixed32(value.data() + value.size() - kTSLength);
+    if ((timestamp_value + ttl) < curtime) {
+      return true; // Data is stale
+    }
+  }
+  return false;
+}
+
+// Strips the TS from the end of the string
+Status DBWithTTL::StripTS(std::string* str) {
+  Status st;
+  // Erasing characters which hold the TS
+  str->erase(str->length() - kTSLength, kTSLength);
+  return st;
+}
+
+Status DBWithTTL::Put(
+    const WriteOptions& o,
+    const Slice& key,
+    const Slice& val) {
+  std::string value_with_ts;
+  Status st = AppendTS(val, value_with_ts);
+  if (!st.ok()) {
+    return st;
+  }
+  return db_->Put(o, key, value_with_ts);
+}
+
+Status DBWithTTL::Get(const ReadOptions& options,
+                      const Slice& key,
+                      std::string* value) {
+  Status st = db_->Get(options, key, value);
+  if (!st.ok()) {
+    return st;
+  }
+  st = SanityCheckTimestamp(*value);
+  if (!st.ok()) {
+    return st;
+  }
+  return StripTS(value);
+}
+
+Status DBWithTTL::Delete(const WriteOptions& wopts, const Slice& key) {
+  return db_->Delete(wopts, key);
+}
+
+Status DBWithTTL::Write(const WriteOptions& opts, WriteBatch* updates) {
+  return db_->Write(opts, updates);
+}
+
+Iterator* DBWithTTL::NewIterator(const ReadOptions& opts) {
+  return new TtlIterator(db_->NewIterator(opts), kTSLength);
+}
+
+const Snapshot* DBWithTTL::GetSnapshot() {
+  return db_->GetSnapshot();
+}
+
+void DBWithTTL::ReleaseSnapshot(const Snapshot* snapshot) {
+  db_->ReleaseSnapshot(snapshot);
+}
+
+bool DBWithTTL::GetProperty(const Slice& property, std::string* value) {
+  return db_->GetProperty(property, value);
+}
+
+void DBWithTTL::GetApproximateSizes(const Range* r, int n, uint64_t* sizes) {
+  db_->GetApproximateSizes(r, n, sizes);
+}
+
+void DBWithTTL::CompactRange(const Slice* begin, const Slice* end) {
+  db_->CompactRange(begin, end);
+}
+
+int DBWithTTL::NumberLevels() {
+  return db_->NumberLevels();
+}
+
+int DBWithTTL::MaxMemCompactionLevel() {
+  return db_->MaxMemCompactionLevel();
+}
+
+int DBWithTTL::Level0StopWriteTrigger() {
+  return db_->Level0StopWriteTrigger();
+}
+
+Status DBWithTTL::Flush(const FlushOptions& fopts) {
+  return db_->Flush(fopts);
+}
+
+Status DBWithTTL::DisableFileDeletions() {
+  return db_->DisableFileDeletions();
+}
+
+Status DBWithTTL::EnableFileDeletions() {
+  return db_->EnableFileDeletions();
+}
+
+Status DBWithTTL::GetLiveFiles(std::vector<std::string>& vec, uint64_t* mfs) {
+  return db_->GetLiveFiles(vec, mfs);
+}
+
+SequenceNumber DBWithTTL::GetLatestSequenceNumber() {
+  return db_->GetLatestSequenceNumber();
+}
+
+Status DBWithTTL::GetUpdatesSince(
+    SequenceNumber seq_number,
+    unique_ptr<TransactionLogIterator>* iter) {
+  return db_->GetUpdatesSince(seq_number, iter);
+}
+
+void DBWithTTL::TEST_Destroy_DBWithTtl() {
+  ((DBImpl*) db_)->TEST_Destroy_DBImpl();
+}
+
+}  // namespace leveldb

utilities/ttl/db_ttl.h

@@ -0,0 +1,95 @@
+// Copyright (c) 2011 The LevelDB Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file. See the AUTHORS file for names of contributors.
+
+#ifndef LEVELDB_UTILITIES_TTL_DB_TTL_H_
+#define LEVELDB_UTILITIES_TTL_DB_TTL_H_
+
+#include "include/leveldb/db.h"
+#include "db/db_impl.h"
+
+namespace leveldb {
+
+class DBWithTTL : public DB {
+ public:
+  DBWithTTL(const int32_t ttl,
+            const Options& options,
+            const std::string& dbname,
+            Status& st);
+
+  virtual ~DBWithTTL();
+
+  virtual Status Put(const WriteOptions& o,
+                     const Slice& key,
+                     const Slice& val);
+
+  virtual Status Get(const ReadOptions& options,
+                     const Slice& key,
+                     std::string* value);
+
+  virtual Status Delete(const WriteOptions& wopts, const Slice& key);
+
+  virtual Status Write(const WriteOptions& opts, WriteBatch* updates);
+
+  virtual Iterator* NewIterator(const ReadOptions& opts);
+
+  virtual const Snapshot* GetSnapshot();
+
+  virtual void ReleaseSnapshot(const Snapshot* snapshot);
+
+  virtual bool GetProperty(const Slice& property, std::string* value);
+
+  virtual void GetApproximateSizes(const Range* r, int n, uint64_t* sizes);
+
+  virtual void CompactRange(const Slice* begin, const Slice* end);
+
+  virtual int NumberLevels();
+
+  virtual int MaxMemCompactionLevel();
+
+  virtual int Level0StopWriteTrigger();
+
+  virtual Status Flush(const FlushOptions& fopts);
+
+  virtual Status DisableFileDeletions();
+
+  virtual Status EnableFileDeletions();
+
+  virtual Status GetLiveFiles(std::vector<std::string>& vec, uint64_t* mfs);
+
+  virtual SequenceNumber GetLatestSequenceNumber();
+
+  virtual Status GetUpdatesSince(SequenceNumber seq_number,
+                                 unique_ptr<TransactionLogIterator>* iter);
+
+  // Simulate a db crash, no elegant closing of database.
+  void TEST_Destroy_DBWithTtl();
+
+  static bool DeleteByTS(void* args,
+                         int level,
+                         const Slice& key,
+                         const Slice& old_val,
+                         std::string* new_val,
+                         bool* value_changed);
+
+  static bool IsStale(const Slice& value, int32_t ttl);
+
+  static Status AppendTS(const Slice& val, std::string& val_with_ts);
+
+  static Status SanityCheckTimestamp(const std::string& str);
+
+  static Status StripTS(std::string* str);
+
+  static Status GetCurrentTime(int32_t& curtime);
+
+  static const int32_t kTSLength = sizeof(int32_t); // size of timestamp
+
+  static const int32_t kMinTimestamp = 1368146402; // 05/09/2013:5:40PM
+
+ private:
+  DB* db_;
+  int32_t ttl_;
+};
+
+}
+#endif  // LEVELDB_UTILITIES_TTL_DB_TTL_H_

utilities/ttl/ttl_test.cc

@@ -0,0 +1,281 @@
+// Copyright (c) 2011 The LevelDB Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file. See the AUTHORS file for names of contributors.
+
+#include "include/utilities/utility_db.h"
+#include "util/testharness.h"
+#include "util/logging.h"
+#include <map>
+#include <unistd.h>
+
+namespace leveldb {
+
+namespace {
+typedef std::map<std::string, std::string> KVMap;
+}
+
+class TtlTest {
+ public:
+  TtlTest() {
+    dbname_ = test::TmpDir() + "/db_ttl";
+    options_.create_if_missing = true;
+    // ensure that compaction is kicked in to always strip timestamp from kvs
+    options_.max_grandparent_overlap_factor = 0;
+    // compaction should take place always from level0 for determinism
+    options_.max_mem_compaction_level = 0;
+    db_ttl_ = nullptr;
+    DestroyDB(dbname_, Options());
+  }
+
+  ~TtlTest() {
+    CloseTtl();
+    DestroyDB(dbname_, Options());
+  }
+
+  // Open database with TTL support when TTL not provided with db_ttl_ pointer
+  void OpenTtl() {
+    assert(db_ttl_ == nullptr); //  db should be closed before opening again
+    ASSERT_OK(UtilityDB::OpenTtlDB(options_, dbname_, &db_ttl_));
+  }
+
+  // Open database with TTL support when TTL provided with db_ttl_ pointer
+  void OpenTtl(int32_t ttl) {
+    assert(db_ttl_ == nullptr);
+    ASSERT_OK(UtilityDB::OpenTtlDB(options_, dbname_, &db_ttl_, ttl));
+  }
+
+  void CloseTtl() {
+    delete db_ttl_;
+    db_ttl_ = nullptr;
+  }
+
+  // Populates and returns a kv-map
+  void MakeKVMap(int64_t num_entries) {
+    kvmap_.clear();
+
+    for (int64_t i = 0; i < num_entries; i++) {
+      std::string key = "key";
+      std::string value = "value";
+      AppendNumberTo(&key, i);
+      AppendNumberTo(&value, i);
+      kvmap_[key] = value;
+    }
+    ASSERT_EQ((int)kvmap_.size(), num_entries);//check all insertions done
+  }
+
+  // Puts num_entries starting from start_pos_map from kvmap_ into the database
+  void PutValues(int start_pos_map, int num_entries) {
+    assert(db_ttl_);
+    ASSERT_LE(start_pos_map + num_entries, (int)kvmap_.size());
+    static WriteOptions wopts;
+    static FlushOptions flush_opts;
+    kv_it_ = kvmap_.begin();
+    advance(kv_it_, start_pos_map);
+    for (int i = 0; kv_it_ != kvmap_.end(), i < num_entries; i++, kv_it_++) {
+      ASSERT_OK(db_ttl_->Put(wopts, kv_it_->first, kv_it_->second));
+    }
+    // Put a mock kv at the end because CompactionFilter doesn't delete last key
+    ASSERT_OK(db_ttl_->Put(wopts, "keymock", "valuemock"));
+    db_ttl_->Flush(flush_opts);
+  }
+
+  // Runs a manual compaction
+  void ManualCompact() {
+    db_ttl_->CompactRange(nullptr, nullptr);
+  }
+
+  // Sleeps for slp_tim then runs a manual compaction
+  // Checks span starting from st_pos from kvmap_ in the db and
+  // Gets should return true if check is true and false otherwise
+  // Also checks that value that we got is the same as inserted
+  void SleepCompactCheck(int slp_tim, int st_pos, int span, bool check = true) {
+    assert(db_ttl_);
+    sleep(slp_tim);
+    ManualCompact();
+    static ReadOptions ropts;
+    kv_it_ = kvmap_.begin();
+    advance(kv_it_, st_pos);
+    std::string v;
+    for (int i = 0; kv_it_ != kvmap_.end(), i < span; i++, kv_it_++) {
+      Status s = db_ttl_->Get(ropts, kv_it_->first, &v);
+      if (s.ok() != check) {
+        fprintf(stderr, "key=%s ",
+                kv_it_->first.c_str());
+        if (!s.ok()) {
+          fprintf(stderr, "is absent from db but was expected to be present\n");
+        } else {
+          fprintf(stderr, "is present in db but was expected to be absent\n");
+        }
+        assert(false);
+      } else if (s.ok() && (v.compare(kv_it_->second) != 0)) {
+          fprintf(stderr, " value for key=%s present in database is %s but "
+                          " should be %s\n", kv_it_->first.c_str(), v.c_str(),
+                          kv_it_->second.c_str());
+          assert(false);
+      }
+    }
+  }
+
+  // Similar as SleepCompactCheck but uses TtlIterator to read from db
+  void SleepCompactCheckIter(int slp, int st_pos, int span, bool check=true) {
+    assert(db_ttl_);
+    sleep(slp);
+    ManualCompact();
+    static ReadOptions ropts;
+    Iterator *dbiter = db_ttl_->NewIterator(ropts);
+    kv_it_ = kvmap_.begin();
+    advance(kv_it_, st_pos);
+
+    dbiter->Seek(kv_it_->first);
+    if (!check) {
+      if (dbiter->Valid()) {
+        ASSERT_NE(dbiter->value().compare(kv_it_->second), 0);
+      }
+    } else {  // dbiter should have found out kvmap_[st_pos]
+      for (int i = st_pos;
+           kv_it_ != kvmap_.end() && i < st_pos + span;
+           i++, kv_it_++)  {
+        ASSERT_TRUE(dbiter->Valid());
+        ASSERT_EQ(dbiter->value().compare(kv_it_->second), 0);
+        dbiter->Next();
+      }
+    }
+    delete dbiter;
+  }
+
+  // Choose carefully so that Put, Gets & Compaction complete in 1 second buffer
+  const int64_t kSampleSize = 100;
+
+ private:
+  std::string dbname_;
+  DB* db_ttl_;
+  Options options_;
+  KVMap kvmap_;
+  KVMap::iterator kv_it_;
+}; // class TtlTest
+
+// If TTL is non positive or not provided, the behaviour is TTL = infinity
+// This test opens the db 3 times with such default behavior and inserts a
+// bunch of kvs each time. All kvs should accummulate in the db till the end
+// Partitions the sample-size provided into 3 sets over boundary1 and boundary2
+TEST(TtlTest, NoEffect) {
+  MakeKVMap(kSampleSize);
+  int boundary1 = kSampleSize / 3;
+  int boundary2 = 2 * boundary1;
+
+  OpenTtl();
+  PutValues(0, boundary1);                       //T=0: Set1 never deleted
+  SleepCompactCheck(1, 0, boundary1);            //T=1: Set1 still there
+  CloseTtl();
+
+  OpenTtl(0);
+  PutValues(boundary1, boundary2 - boundary1);   //T=1: Set2 never deleted
+  SleepCompactCheck(1, 0, boundary2);            //T=2: Sets1 & 2 still there
+  CloseTtl();
+
+  OpenTtl(-1);
+  PutValues(boundary2, kSampleSize - boundary2); //T=3: Set3 never deleted
+  SleepCompactCheck(1, 0, kSampleSize, true);    //T=4: Sets 1,2,3 still there
+  CloseTtl();
+}
+
+// Puts a set of values and checks its presence using Get during ttl
+TEST(TtlTest, PresentDuringTTL) {
+  MakeKVMap(kSampleSize);
+
+  OpenTtl(2);                                 // T=0:Open the db with ttl = 2
+  PutValues(0, kSampleSize);                  // T=0:Insert Set1. Delete at t=2
+  SleepCompactCheck(1, 0, kSampleSize, true); // T=1:Set1 should still be there
+  CloseTtl();
+}
+
+// Puts a set of values and checks its absence using Get after ttl
+TEST(TtlTest, AbsentAfterTTL) {
+  MakeKVMap(kSampleSize);
+
+  OpenTtl(1);                                  // T=0:Open the db with ttl = 2
+  PutValues(0, kSampleSize);                   // T=0:Insert Set1. Delete at t=2
+  SleepCompactCheck(2, 0, kSampleSize, false); // T=2:Set1 should not be there
+  CloseTtl();
+}
+
+// Resets the timestamp of a set of kvs by updating them and checks that they
+// are not deleted according to the old timestamp
+TEST(TtlTest, ResetTimestamp) {
+  MakeKVMap(kSampleSize);
+
+  OpenTtl(3);
+  PutValues(0, kSampleSize);            // T=0: Insert Set1. Delete at t=3
+  sleep(2);                             // T=2
+  PutValues(0, kSampleSize);            // T=2: Insert Set1. Delete at t=5
+  SleepCompactCheck(2, 0, kSampleSize); // T=4: Set1 should still be there
+  CloseTtl();
+}
+
+// Similar to PresentDuringTTL but uses Iterator
+TEST(TtlTest, IterPresentDuringTTL) {
+  MakeKVMap(kSampleSize);
+
+  OpenTtl(2);
+  PutValues(0, kSampleSize);                 // T=0: Insert. Delete at t=2
+  SleepCompactCheckIter(1, 0, kSampleSize);  // T=1: Set should be there
+  CloseTtl();
+}
+
+// Similar to AbsentAfterTTL but uses Iterator
+TEST(TtlTest, IterAbsentAfterTTL) {
+  MakeKVMap(kSampleSize);
+
+  OpenTtl(1);
+  PutValues(0, kSampleSize);                       // T=0: Insert. Delete at t=1
+  SleepCompactCheckIter(2, 0, kSampleSize, false); // T=2: Should not be there
+  CloseTtl();
+}
+
+// Checks presence while opening the same db more than once with the same ttl
+// Note: The second open will open the same db
+TEST(TtlTest, MultiOpenSamePresent) {
+  MakeKVMap(kSampleSize);
+
+  OpenTtl(2);
+  PutValues(0, kSampleSize);                   // T=0: Insert. Delete at t=2
+  CloseTtl();
+
+  OpenTtl(2);                                  // T=0. Delete at t=2
+  SleepCompactCheck(1, 0, kSampleSize);        // T=1: Set should be there
+  CloseTtl();
+}
+
+// Checks absence while opening the same db more than once with the same ttl
+// Note: The second open will open the same db
+TEST(TtlTest, MultiOpenSameAbsent) {
+  MakeKVMap(kSampleSize);
+
+  OpenTtl(1);
+  PutValues(0, kSampleSize);                   // T=0: Insert. Delete at t=1
+  CloseTtl();
+
+  OpenTtl(1);                                  // T=0.Delete at t=1
+  SleepCompactCheck(2, 0, kSampleSize, false); // T=2: Set should not be there
+  CloseTtl();
+}
+
+// Checks presence while opening the same db more than once with bigger ttl
+TEST(TtlTest, MultiOpenDifferent) {
+  MakeKVMap(kSampleSize);
+
+  OpenTtl(1);
+  PutValues(0, kSampleSize);            // T=0: Insert. Delete at t=1
+  CloseTtl();
+
+  OpenTtl(3);                           // T=0: Set deleted at t=3
+  SleepCompactCheck(2, 0, kSampleSize); // T=2: Set should be there
+  CloseTtl();
+}
+
+} //  namespace leveldb
+
+// A black-box test for the ttl wrapper around rocksdb
+int main(int argc, char** argv) {
+  return leveldb::test::RunAllTests();
+}