Timestamp and TTL Wrapper for rocksdb

Summary:
When opened with DBTimestamp::Open call, timestamps are prepended to and stripped from the value during subsequent Put and Get calls respectively. The Timestamp is used to discard values in Get and custom compaction filter which have exceeded their TTL which is specified during Open.
Have made a temporary change to Makefile to let us test with the temporary file TestTime.cc. Have also changed the private members of db_impl.h to protected to let them be inherited by the new class DBTimestamp

Test Plan: make db_timestamp; TestTime.cc(will not check it in) shows how to use the apis currently, but I will write unit-tests shortly

Reviewers: dhruba, vamsi, haobo, sheki, heyongqiang, vkrest

Reviewed By: vamsi

CC: zshao, xjin, vkrest, MarkCallaghan

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

utilities/ttl/db_ttl.cc

@@ -0,0 +1,256 @@
+// 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(iter_->value().size() >= (unsigned)ts_len_);
+    return std::string(iter_->value().data(), iter_->value().size() - ts_len_);
+  }
+
+  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;
+}
+
+// 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;
+  if (str->length() < (unsigned)kTSLength) {
+    return Status::IOError("Error: value's length less than timestamp's\n");
+  }
+  // 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;
+  }
+  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,91 @@
+// 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 StripTS(std::string* str);
+
+  static Status GetCurrentTime(int32_t& curtime);
+
+  static const int32_t kTSLength = sizeof(int32_t); // size of timestamp
+
+ 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();
+}