support stackableDB as the baseDB of transactionDB

Summary: make transactionDB working with StackableDB

Test Plan: make all check -j64

Reviewers: andrewkr, yiwu, sdong

Reviewed By: sdong

Subscribers: andrewkr, dhruba, leveldb

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

include/rocksdb/utilities/transaction_db.h

@@ -98,6 +98,7 @@ struct TransactionOptions {
 class TransactionDB : public StackableDB {
  public:
   // Open a TransactionDB similar to DB::Open().
+  // Internally call PrepareWrap() and WrapDB()
   static Status Open(const Options& options,
                      const TransactionDBOptions& txn_db_options,
                      const std::string& dbname, TransactionDB** dbptr);
@@ -108,7 +109,27 @@ class TransactionDB : public StackableDB {
                      const std::vector<ColumnFamilyDescriptor>& column_families,
                      std::vector<ColumnFamilyHandle*>* handles,
                      TransactionDB** dbptr);
-
+  // The following functions are used to open a TransactionDB internally using
+  // an opened DB or StackableDB.
+  // 1. Call prepareWrap(), passing an empty std::vector<size_t> to
+  // compaction_enabled_cf_indices.
+  // 2. Open DB or Stackable DB with db_options and column_families passed to
+  // prepareWrap()
+  // Note: PrepareWrap() may change parameters, make copies before the
+  // invocation if needed.
+  // 3. Call Wrap*DB() with compaction_enabled_cf_indices in step 1 and handles
+  // of the opened DB/StackableDB in step 2
+  static void PrepareWrap(DBOptions* db_options,
+                          std::vector<ColumnFamilyDescriptor>* column_families,
+                          std::vector<size_t>* compaction_enabled_cf_indices);
+  static Status WrapDB(DB* db, const TransactionDBOptions& txn_db_options,
+                       const std::vector<size_t>& compaction_enabled_cf_indices,
+                       const std::vector<ColumnFamilyHandle*>& handles,
+                       TransactionDB** dbptr);
+  static Status WrapStackableDB(
+      StackableDB* db, const TransactionDBOptions& txn_db_options,
+      const std::vector<size_t>& compaction_enabled_cf_indices,
+      const std::vector<ColumnFamilyHandle*>& handles, TransactionDB** dbptr);
   virtual ~TransactionDB() {}
 
   // Starts a new Transaction.

utilities/transactions/transaction_db_impl.cc

@@ -33,12 +33,93 @@ TransactionDBImpl::TransactionDBImpl(DB* db,
   assert(db_impl_ != nullptr);
 }
 
+// Support initiliazing TransactionDBImpl from a stackable db
+//
+//    TransactionDBImpl
+//     ^        ^
+//     |        |
+//     |        +
+//     |   StackableDB
+//     |   ^
+//     |   |
+//     +   +
+//     DBImpl
+//       ^
+//       |(inherit)
+//       +
+//       DB
+//
+TransactionDBImpl::TransactionDBImpl(StackableDB* db,
+                                     const TransactionDBOptions& txn_db_options)
+    : TransactionDB(db),
+      db_impl_(dynamic_cast<DBImpl*>(db->GetRootDB())),
+      txn_db_options_(txn_db_options),
+      lock_mgr_(this, txn_db_options_.num_stripes, txn_db_options.max_num_locks,
+                txn_db_options_.custom_mutex_factory
+                    ? txn_db_options_.custom_mutex_factory
+                    : std::shared_ptr<TransactionDBMutexFactory>(
+                          new TransactionDBMutexFactoryImpl())) {
+  assert(db_impl_ != nullptr);
+}
+
 TransactionDBImpl::~TransactionDBImpl() {
   while (!transactions_.empty()) {
     delete transactions_.begin()->second;
   }
 }
 
+Status TransactionDBImpl::Initialize(
+    const std::vector<size_t>& compaction_enabled_cf_indices,
+    const std::vector<ColumnFamilyHandle*>& handles) {
+  for (auto cf_ptr : handles) {
+    AddColumnFamily(cf_ptr);
+  }
+  // Re-enable compaction for the column families that initially had
+  // compaction enabled.
+  std::vector<ColumnFamilyHandle*> compaction_enabled_cf_handles;
+  compaction_enabled_cf_handles.reserve(compaction_enabled_cf_indices.size());
+  for (auto index : compaction_enabled_cf_indices) {
+    compaction_enabled_cf_handles.push_back(handles[index]);
+  }
+
+  Status s = EnableAutoCompaction(compaction_enabled_cf_handles);
+
+  // create 'real' transactions from recovered shell transactions
+  auto dbimpl = reinterpret_cast<DBImpl*>(GetRootDB());
+  assert(dbimpl != nullptr);
+  auto rtrxs = dbimpl->recovered_transactions();
+
+  for (auto it = rtrxs.begin(); it != rtrxs.end(); it++) {
+    auto recovered_trx = it->second;
+    assert(recovered_trx);
+    assert(recovered_trx->log_number_);
+    assert(recovered_trx->name_.length());
+
+    WriteOptions w_options;
+    w_options.sync = true;
+    TransactionOptions t_options;
+
+    Transaction* real_trx = BeginTransaction(w_options, t_options, nullptr);
+    assert(real_trx);
+    real_trx->SetLogNumber(recovered_trx->log_number_);
+
+    s = real_trx->SetName(recovered_trx->name_);
+    if (!s.ok()) {
+      break;
+    }
+
+    s = real_trx->RebuildFromWriteBatch(recovered_trx->batch_);
+    real_trx->exec_status_ = Transaction::PREPARED;
+    if (!s.ok()) {
+      break;
+    }
+  }
+  if (s.ok()) {
+    dbimpl->DeleteAllRecoveredTransactions();
+  }
+  return s;
+}
+
 Transaction* TransactionDBImpl::BeginTransaction(
     const WriteOptions& write_options, const TransactionOptions& txn_options,
     Transaction* old_txn) {
@@ -92,83 +173,63 @@ Status TransactionDB::Open(
 
   std::vector<ColumnFamilyDescriptor> column_families_copy = column_families;
   std::vector<size_t> compaction_enabled_cf_indices;
-
-  // Enable MemTable History if not already enabled
-  for (size_t i = 0; i < column_families_copy.size(); i++) {
-    ColumnFamilyOptions* options = &column_families_copy[i].options;
-
-    if (options->max_write_buffer_number_to_maintain == 0) {
-      // Setting to -1 will set the History size to max_write_buffer_number.
-      options->max_write_buffer_number_to_maintain = -1;
-    }
-
-    if (!options->disable_auto_compactions) {
-      // Disable compactions momentarily to prevent race with DB::Open
-      options->disable_auto_compactions = true;
-      compaction_enabled_cf_indices.push_back(i);
-    }
-  }
-
   DBOptions db_options_2pc = db_options;
-  db_options_2pc.allow_2pc = true;
+  PrepareWrap(&db_options_2pc, &column_families_copy,
+              &compaction_enabled_cf_indices);
   s = DB::Open(db_options_2pc, dbname, column_families_copy, handles, &db);
-
   if (s.ok()) {
-    TransactionDBImpl* txn_db = new TransactionDBImpl(
-        db, TransactionDBImpl::ValidateTxnDBOptions(txn_db_options));
-    *dbptr = txn_db;
+    s = WrapDB(db, txn_db_options, compaction_enabled_cf_indices, *handles,
+               dbptr);
+  }
+  return s;
+}
 
-    for (auto cf_ptr : *handles) {
-      txn_db->AddColumnFamily(cf_ptr);
-    }
+void TransactionDB::PrepareWrap(
+    DBOptions* db_options, std::vector<ColumnFamilyDescriptor>* column_families,
+    std::vector<size_t>* compaction_enabled_cf_indices) {
+  compaction_enabled_cf_indices->clear();
 
-    // Re-enable compaction for the column families that initially had
-    // compaction enabled.
-    assert(column_families_copy.size() == (*handles).size());
-    std::vector<ColumnFamilyHandle*> compaction_enabled_cf_handles;
-    compaction_enabled_cf_handles.reserve(compaction_enabled_cf_indices.size());
-    for (auto index : compaction_enabled_cf_indices) {
-      compaction_enabled_cf_handles.push_back((*handles)[index]);
-    }
+  // Enable MemTable History if not already enabled
+  for (size_t i = 0; i < column_families->size(); i++) {
+    ColumnFamilyOptions* cf_options = &(*column_families)[i].options;
 
-    s = txn_db->EnableAutoCompaction(compaction_enabled_cf_handles);
-
-    // create 'real' transactions from recovered shell transactions
-    assert(dynamic_cast<DBImpl*>(db) != nullptr);
-    auto dbimpl = reinterpret_cast<DBImpl*>(db);
-    auto rtrxs = dbimpl->recovered_transactions();
-
-    for (auto it = rtrxs.begin(); it != rtrxs.end(); it++) {
-      auto recovered_trx = it->second;
-      assert(recovered_trx);
-      assert(recovered_trx->log_number_);
-      assert(recovered_trx->name_.length());
-
-      WriteOptions w_options;
-      w_options.sync = true;
-      TransactionOptions t_options;
-
-      Transaction* real_trx =
-          txn_db->BeginTransaction(w_options, t_options, nullptr);
-      assert(real_trx);
-      real_trx->SetLogNumber(recovered_trx->log_number_);
-
-      s = real_trx->SetName(recovered_trx->name_);
-      if (!s.ok()) {
-        break;
-      }
-
-      s = real_trx->RebuildFromWriteBatch(recovered_trx->batch_);
-      real_trx->exec_status_ = Transaction::PREPARED;
-      if (!s.ok()) {
-        break;
-      }
+    if (cf_options->max_write_buffer_number_to_maintain == 0) {
+      // Setting to -1 will set the History size to max_write_buffer_number.
+      cf_options->max_write_buffer_number_to_maintain = -1;
     }
-    if (s.ok()) {
-      dbimpl->DeleteAllRecoveredTransactions();
+    if (!cf_options->disable_auto_compactions) {
+      // Disable compactions momentarily to prevent race with DB::Open
+      cf_options->disable_auto_compactions = true;
+      compaction_enabled_cf_indices->push_back(i);
     }
   }
+  db_options->allow_2pc = true;
+}
+
+Status TransactionDB::WrapDB(
+    // make sure this db is already opened with memtable history enabled,
+    // auto compaction distabled and 2 phase commit enabled
+    DB* db, const TransactionDBOptions& txn_db_options,
+    const std::vector<size_t>& compaction_enabled_cf_indices,
+    const std::vector<ColumnFamilyHandle*>& handles, TransactionDB** dbptr) {
+  TransactionDBImpl* txn_db = new TransactionDBImpl(
+      db, TransactionDBImpl::ValidateTxnDBOptions(txn_db_options));
+  *dbptr = txn_db;
+  Status s = txn_db->Initialize(compaction_enabled_cf_indices, handles);
+  return s;
+}
 
+Status TransactionDB::WrapStackableDB(
+    // make sure this stackable_db is already opened with memtable history
+    // enabled,
+    // auto compaction distabled and 2 phase commit enabled
+    StackableDB* db, const TransactionDBOptions& txn_db_options,
+    const std::vector<size_t>& compaction_enabled_cf_indices,
+    const std::vector<ColumnFamilyHandle*>& handles, TransactionDB** dbptr) {
+  TransactionDBImpl* txn_db = new TransactionDBImpl(
+      db, TransactionDBImpl::ValidateTxnDBOptions(txn_db_options));
+  *dbptr = txn_db;
+  Status s = txn_db->Initialize(compaction_enabled_cf_indices, handles);
   return s;
 }
 

utilities/transactions/transaction_db_impl.h

@@ -25,8 +25,14 @@ class TransactionDBImpl : public TransactionDB {
   explicit TransactionDBImpl(DB* db,
                              const TransactionDBOptions& txn_db_options);
 
+  explicit TransactionDBImpl(StackableDB* db,
+                             const TransactionDBOptions& txn_db_options);
+
   ~TransactionDBImpl();
 
+  Status Initialize(const std::vector<size_t>& compaction_enabled_cf_indices,
+                    const std::vector<ColumnFamilyHandle*>& handles);
+
   Transaction* BeginTransaction(const WriteOptions& write_options,
                                 const TransactionOptions& txn_options,
                                 Transaction* old_txn) override;

utilities/transactions/transaction_impl.cc

@@ -37,17 +37,15 @@ TransactionID TransactionImpl::GenTxnID() {
 TransactionImpl::TransactionImpl(TransactionDB* txn_db,
                                  const WriteOptions& write_options,
                                  const TransactionOptions& txn_options)
-    : TransactionBaseImpl(txn_db->GetBaseDB(), write_options),
+    : TransactionBaseImpl(txn_db->GetRootDB(), write_options),
       txn_db_impl_(nullptr),
       txn_id_(0),
       expiration_time_(0),
       lock_timeout_(0) {
   txn_db_impl_ = dynamic_cast<TransactionDBImpl*>(txn_db);
   assert(txn_db_impl_);
-
-  db_impl_ = dynamic_cast<DBImpl*>(txn_db->GetBaseDB());
+  db_impl_ = dynamic_cast<DBImpl*>(txn_db->GetRootDB());
   assert(db_impl_);
-
   Initialize(txn_options);
 }
 
@@ -99,7 +97,7 @@ void TransactionImpl::Reinitialize(TransactionDB* txn_db,
   if (!name_.empty() && exec_status_ != COMMITED) {
     txn_db_impl_->UnregisterTransaction(this);
   }
-  TransactionBaseImpl::Reinitialize(txn_db->GetBaseDB(), write_options);
+  TransactionBaseImpl::Reinitialize(txn_db->GetRootDB(), write_options);
   Initialize(txn_options);
 }
 

utilities/transactions/transaction_test.cc

@@ -28,7 +28,7 @@ using std::string;
 
 namespace rocksdb {
 
-class TransactionTest : public testing::Test {
+class TransactionTest : public ::testing::TestWithParam<bool> {
  public:
   TransactionDB* db;
   FaultInjectionTestEnv* env;
@@ -50,7 +50,12 @@ class TransactionTest : public testing::Test {
     DestroyDB(dbname, options);
     txn_db_options.transaction_lock_timeout = 0;
     txn_db_options.default_lock_timeout = 0;
-    Status s = TransactionDB::Open(options, txn_db_options, dbname, &db);
+    Status s;
+    if (GetParam() == false) {
+      s = TransactionDB::Open(options, txn_db_options, dbname, &db);
+    } else {
+      s = OpenWithStackableDB();
+    }
     assert(s.ok());
   }
 
@@ -66,21 +71,55 @@ class TransactionTest : public testing::Test {
     env->AssertNoOpenFile();
     env->DropUnsyncedFileData();
     env->ResetState();
-    Status s = TransactionDB::Open(options, txn_db_options, dbname, &db);
+    Status s;
+    if (GetParam() == false) {
+      s = TransactionDB::Open(options, txn_db_options, dbname, &db);
+    } else {
+      s = OpenWithStackableDB();
+    }
     return s;
   }
 
   Status ReOpen() {
     delete db;
     DestroyDB(dbname, options);
+    Status s;
+    if (GetParam() == false) {
+      s = TransactionDB::Open(options, txn_db_options, dbname, &db);
+    } else {
+      s = OpenWithStackableDB();
+    }
+    return s;
+  }
 
-    Status s = TransactionDB::Open(options, txn_db_options, dbname, &db);
-
+  Status OpenWithStackableDB() {
+    std::vector<size_t> compaction_enabled_cf_indices;
+    std::vector<ColumnFamilyDescriptor> column_families{ColumnFamilyDescriptor(
+        kDefaultColumnFamilyName, ColumnFamilyOptions(options))};
+
+    TransactionDB::PrepareWrap(&options, &column_families,
+                               &compaction_enabled_cf_indices);
+    std::vector<ColumnFamilyHandle*> handles;
+    DB* root_db;
+    Options options_copy(options);
+    Status s =
+        DB::Open(options_copy, dbname, column_families, &handles, &root_db);
+    if (s.ok()) {
+      assert(handles.size() == 1);
+      s = TransactionDB::WrapStackableDB(
+          new StackableDB(root_db), txn_db_options,
+          compaction_enabled_cf_indices, handles, &db);
+      delete handles[0];
+    }
     return s;
   }
 };
 
-TEST_F(TransactionTest, DoubleEmptyWrite) {
+INSTANTIATE_TEST_CASE_P(DBAsBaseDB, TransactionTest, ::testing::Values(false));
+INSTANTIATE_TEST_CASE_P(StackableDBAsBaseDB, TransactionTest,
+                        ::testing::Values(true));
+
+TEST_P(TransactionTest, DoubleEmptyWrite) {
   WriteOptions write_options;
   write_options.sync = true;
   write_options.disableWAL = false;
@@ -91,7 +130,7 @@ TEST_F(TransactionTest, DoubleEmptyWrite) {
   ASSERT_OK(db->Write(write_options, &batch));
 }
 
-TEST_F(TransactionTest, SuccessTest) {
+TEST_P(TransactionTest, SuccessTest) {
   WriteOptions write_options;
   ReadOptions read_options;
   string value;
@@ -128,7 +167,7 @@ TEST_F(TransactionTest, SuccessTest) {
   delete txn;
 }
 
-TEST_F(TransactionTest, CommitTimeBatchFailTest) {
+TEST_P(TransactionTest, CommitTimeBatchFailTest) {
   WriteOptions write_options;
   TransactionOptions txn_options;
 
@@ -150,7 +189,7 @@ TEST_F(TransactionTest, CommitTimeBatchFailTest) {
   delete txn1;
 }
 
-TEST_F(TransactionTest, SimpleTwoPhaseTransactionTest) {
+TEST_P(TransactionTest, SimpleTwoPhaseTransactionTest) {
   WriteOptions write_options;
   ReadOptions read_options;
 
@@ -248,7 +287,7 @@ TEST_F(TransactionTest, SimpleTwoPhaseTransactionTest) {
   delete txn;
 }
 
-TEST_F(TransactionTest, TwoPhaseNameTest) {
+TEST_P(TransactionTest, TwoPhaseNameTest) {
   Status s;
 
   WriteOptions write_options;
@@ -305,7 +344,7 @@ TEST_F(TransactionTest, TwoPhaseNameTest) {
   delete txn2;
 }
 
-TEST_F(TransactionTest, TwoPhaseEmptyWriteTest) {
+TEST_P(TransactionTest, TwoPhaseEmptyWriteTest) {
   Status s;
   std::string value;
 
@@ -346,7 +385,7 @@ TEST_F(TransactionTest, TwoPhaseEmptyWriteTest) {
   delete txn2;
 }
 
-TEST_F(TransactionTest, TwoPhaseExpirationTest) {
+TEST_P(TransactionTest, TwoPhaseExpirationTest) {
   Status s;
 
   WriteOptions write_options;
@@ -378,7 +417,7 @@ TEST_F(TransactionTest, TwoPhaseExpirationTest) {
   delete txn2;
 }
 
-TEST_F(TransactionTest, TwoPhaseRollbackTest) {
+TEST_P(TransactionTest, TwoPhaseRollbackTest) {
   WriteOptions write_options;
   ReadOptions read_options;
 
@@ -447,7 +486,7 @@ TEST_F(TransactionTest, TwoPhaseRollbackTest) {
   delete txn;
 }
 
-TEST_F(TransactionTest, PersistentTwoPhaseTransactionTest) {
+TEST_P(TransactionTest, PersistentTwoPhaseTransactionTest) {
   WriteOptions write_options;
   write_options.sync = true;
   write_options.disableWAL = false;
@@ -562,7 +601,7 @@ TEST_F(TransactionTest, PersistentTwoPhaseTransactionTest) {
   ASSERT_EQ(db->GetTransactionByName("xid"), nullptr);
 }
 
-TEST_F(TransactionTest, TwoPhaseMultiThreadTest) {
+TEST_P(TransactionTest, TwoPhaseMultiThreadTest) {
   // mix transaction writes and regular writes
   const uint32_t NUM_TXN_THREADS = 50;
   std::atomic<uint32_t> txn_thread_num(0);
@@ -643,7 +682,7 @@ TEST_F(TransactionTest, TwoPhaseMultiThreadTest) {
   }
 }
 
-TEST_F(TransactionTest, TwoPhaseLongPrepareTest) {
+TEST_P(TransactionTest, TwoPhaseLongPrepareTest) {
   WriteOptions write_options;
   write_options.sync = true;
   write_options.disableWAL = false;
@@ -706,7 +745,7 @@ TEST_F(TransactionTest, TwoPhaseLongPrepareTest) {
   delete txn;
 }
 
-TEST_F(TransactionTest, TwoPhaseSequenceTest) {
+TEST_P(TransactionTest, TwoPhaseSequenceTest) {
   WriteOptions write_options;
   write_options.sync = true;
   write_options.disableWAL = false;
@@ -751,7 +790,7 @@ TEST_F(TransactionTest, TwoPhaseSequenceTest) {
   ASSERT_EQ(value, "bar4");
 }
 
-TEST_F(TransactionTest, TwoPhaseDoubleRecoveryTest) {
+TEST_P(TransactionTest, TwoPhaseDoubleRecoveryTest) {
   WriteOptions write_options;
   write_options.sync = true;
   write_options.disableWAL = false;
@@ -820,7 +859,7 @@ TEST_F(TransactionTest, TwoPhaseDoubleRecoveryTest) {
   ASSERT_EQ(value, "bar2");
 }
 
-TEST_F(TransactionTest, TwoPhaseLogRollingTest) {
+TEST_P(TransactionTest, TwoPhaseLogRollingTest) {
   DBImpl* db_impl = reinterpret_cast<DBImpl*>(db->GetRootDB());
 
   Status s;
@@ -939,7 +978,7 @@ TEST_F(TransactionTest, TwoPhaseLogRollingTest) {
   delete cfb;
 }
 
-TEST_F(TransactionTest, FirstWriteTest) {
+TEST_P(TransactionTest, FirstWriteTest) {
   WriteOptions write_options;
 
   // Test conflict checking against the very first write to a db.
@@ -958,7 +997,7 @@ TEST_F(TransactionTest, FirstWriteTest) {
   delete txn;
 }
 
-TEST_F(TransactionTest, FirstWriteTest2) {
+TEST_P(TransactionTest, FirstWriteTest2) {
   WriteOptions write_options;
 
   Transaction* txn = db->BeginTransaction(write_options);
@@ -976,7 +1015,7 @@ TEST_F(TransactionTest, FirstWriteTest2) {
   delete txn;
 }
 
-TEST_F(TransactionTest, WriteOptionsTest) {
+TEST_P(TransactionTest, WriteOptionsTest) {
   WriteOptions write_options;
   write_options.sync = true;
   write_options.disableWAL = true;
@@ -994,7 +1033,7 @@ TEST_F(TransactionTest, WriteOptionsTest) {
   delete txn;
 }
 
-TEST_F(TransactionTest, WriteConflictTest) {
+TEST_P(TransactionTest, WriteConflictTest) {
   WriteOptions write_options;
   ReadOptions read_options;
   string value;
@@ -1030,7 +1069,7 @@ TEST_F(TransactionTest, WriteConflictTest) {
   delete txn;
 }
 
-TEST_F(TransactionTest, WriteConflictTest2) {
+TEST_P(TransactionTest, WriteConflictTest2) {
   WriteOptions write_options;
   ReadOptions read_options;
   TransactionOptions txn_options;
@@ -1078,7 +1117,7 @@ TEST_F(TransactionTest, WriteConflictTest2) {
   delete txn;
 }
 
-TEST_F(TransactionTest, ReadConflictTest) {
+TEST_P(TransactionTest, ReadConflictTest) {
   WriteOptions write_options;
   ReadOptions read_options, snapshot_read_options;
   TransactionOptions txn_options;
@@ -1114,7 +1153,7 @@ TEST_F(TransactionTest, ReadConflictTest) {
   delete txn;
 }
 
-TEST_F(TransactionTest, TxnOnlyTest) {
+TEST_P(TransactionTest, TxnOnlyTest) {
   // Test to make sure transactions work when there are no other writes in an
   // empty db.
 
@@ -1135,7 +1174,7 @@ TEST_F(TransactionTest, TxnOnlyTest) {
   delete txn;
 }
 
-TEST_F(TransactionTest, FlushTest) {
+TEST_P(TransactionTest, FlushTest) {
   WriteOptions write_options;
   ReadOptions read_options, snapshot_read_options;
   string value;
@@ -1176,7 +1215,7 @@ TEST_F(TransactionTest, FlushTest) {
   delete txn;
 }
 
-TEST_F(TransactionTest, FlushTest2) {
+TEST_P(TransactionTest, FlushTest2) {
   const size_t num_tests = 3;
 
   for (size_t n = 0; n < num_tests; n++) {
@@ -1203,7 +1242,7 @@ TEST_F(TransactionTest, FlushTest2) {
     TransactionOptions txn_options;
     string value;
 
-    DBImpl* db_impl = reinterpret_cast<DBImpl*>(db->GetBaseDB());
+    DBImpl* db_impl = reinterpret_cast<DBImpl*>(db->GetRootDB());
 
     db->Put(write_options, Slice("foo"), Slice("bar"));
     db->Put(write_options, Slice("foo2"), Slice("bar2"));
@@ -1327,7 +1366,7 @@ TEST_F(TransactionTest, FlushTest2) {
   }
 }
 
-TEST_F(TransactionTest, NoSnapshotTest) {
+TEST_P(TransactionTest, NoSnapshotTest) {
   WriteOptions write_options;
   ReadOptions read_options;
   string value;
@@ -1357,7 +1396,7 @@ TEST_F(TransactionTest, NoSnapshotTest) {
   delete txn;
 }
 
-TEST_F(TransactionTest, MultipleSnapshotTest) {
+TEST_P(TransactionTest, MultipleSnapshotTest) {
   WriteOptions write_options;
   ReadOptions read_options, snapshot_read_options;
   string value;
@@ -1474,7 +1513,7 @@ TEST_F(TransactionTest, MultipleSnapshotTest) {
   delete txn2;
 }
 
-TEST_F(TransactionTest, ColumnFamiliesTest) {
+TEST_P(TransactionTest, ColumnFamiliesTest) {
   WriteOptions write_options;
   ReadOptions read_options, snapshot_read_options;
   TransactionOptions txn_options;
@@ -1641,7 +1680,7 @@ TEST_F(TransactionTest, ColumnFamiliesTest) {
   }
 }
 
-TEST_F(TransactionTest, ColumnFamiliesTest2) {
+TEST_P(TransactionTest, ColumnFamiliesTest2) {
   WriteOptions write_options;
   ReadOptions read_options, snapshot_read_options;
   TransactionOptions txn_options;
@@ -1718,7 +1757,7 @@ TEST_F(TransactionTest, ColumnFamiliesTest2) {
   delete two;
 }
 
-TEST_F(TransactionTest, EmptyTest) {
+TEST_P(TransactionTest, EmptyTest) {
   WriteOptions write_options;
   ReadOptions read_options;
   string value;
@@ -1760,7 +1799,7 @@ TEST_F(TransactionTest, EmptyTest) {
   delete txn;
 }
 
-TEST_F(TransactionTest, PredicateManyPreceders) {
+TEST_P(TransactionTest, PredicateManyPreceders) {
   WriteOptions write_options;
   ReadOptions read_options1, read_options2;
   TransactionOptions txn_options;
@@ -1822,7 +1861,7 @@ TEST_F(TransactionTest, PredicateManyPreceders) {
   delete txn2;
 }
 
-TEST_F(TransactionTest, LostUpdate) {
+TEST_P(TransactionTest, LostUpdate) {
   WriteOptions write_options;
   ReadOptions read_options, read_options1, read_options2;
   TransactionOptions txn_options;
@@ -1948,7 +1987,7 @@ TEST_F(TransactionTest, LostUpdate) {
   ASSERT_EQ(value, "10");
 }
 
-TEST_F(TransactionTest, UntrackedWrites) {
+TEST_P(TransactionTest, UntrackedWrites) {
   WriteOptions write_options;
   ReadOptions read_options;
   string value;
@@ -1992,7 +2031,7 @@ TEST_F(TransactionTest, UntrackedWrites) {
   delete txn;
 }
 
-TEST_F(TransactionTest, ExpiredTransaction) {
+TEST_P(TransactionTest, ExpiredTransaction) {
   WriteOptions write_options;
   ReadOptions read_options;
   TransactionOptions txn_options;
@@ -2038,7 +2077,7 @@ TEST_F(TransactionTest, ExpiredTransaction) {
   delete txn2;
 }
 
-TEST_F(TransactionTest, ReinitializeTest) {
+TEST_P(TransactionTest, ReinitializeTest) {
   WriteOptions write_options;
   ReadOptions read_options;
   TransactionOptions txn_options;
@@ -2144,7 +2183,7 @@ TEST_F(TransactionTest, ReinitializeTest) {
   delete txn1;
 }
 
-TEST_F(TransactionTest, Rollback) {
+TEST_P(TransactionTest, Rollback) {
   WriteOptions write_options;
   ReadOptions read_options;
   TransactionOptions txn_options;
@@ -2181,7 +2220,7 @@ TEST_F(TransactionTest, Rollback) {
   delete txn2;
 }
 
-TEST_F(TransactionTest, LockLimitTest) {
+TEST_P(TransactionTest, LockLimitTest) {
   WriteOptions write_options;
   ReadOptions read_options, snapshot_read_options;
   TransactionOptions txn_options;
@@ -2285,7 +2324,7 @@ TEST_F(TransactionTest, LockLimitTest) {
   delete txn2;
 }
 
-TEST_F(TransactionTest, IteratorTest) {
+TEST_P(TransactionTest, IteratorTest) {
   WriteOptions write_options;
   ReadOptions read_options, snapshot_read_options;
   TransactionOptions txn_options;
@@ -2404,7 +2443,7 @@ TEST_F(TransactionTest, IteratorTest) {
   delete txn;
 }
 
-TEST_F(TransactionTest, DisableIndexingTest) {
+TEST_P(TransactionTest, DisableIndexingTest) {
   WriteOptions write_options;
   ReadOptions read_options;
   string value;
@@ -2466,7 +2505,7 @@ TEST_F(TransactionTest, DisableIndexingTest) {
   delete txn;
 }
 
-TEST_F(TransactionTest, SavepointTest) {
+TEST_P(TransactionTest, SavepointTest) {
   WriteOptions write_options;
   ReadOptions read_options, snapshot_read_options;
   TransactionOptions txn_options;
@@ -2660,7 +2699,7 @@ TEST_F(TransactionTest, SavepointTest) {
   delete txn;
 }
 
-TEST_F(TransactionTest, SavepointTest2) {
+TEST_P(TransactionTest, SavepointTest2) {
   WriteOptions write_options;
   ReadOptions read_options;
   TransactionOptions txn_options;
@@ -2758,7 +2797,7 @@ TEST_F(TransactionTest, SavepointTest2) {
   delete txn2;
 }
 
-TEST_F(TransactionTest, UndoGetForUpdateTest) {
+TEST_P(TransactionTest, UndoGetForUpdateTest) {
   WriteOptions write_options;
   ReadOptions read_options;
   TransactionOptions txn_options;
@@ -2902,7 +2941,7 @@ TEST_F(TransactionTest, UndoGetForUpdateTest) {
   delete txn1;
 }
 
-TEST_F(TransactionTest, UndoGetForUpdateTest2) {
+TEST_P(TransactionTest, UndoGetForUpdateTest2) {
   WriteOptions write_options;
   ReadOptions read_options;
   TransactionOptions txn_options;
@@ -3108,7 +3147,7 @@ TEST_F(TransactionTest, UndoGetForUpdateTest2) {
   delete txn2;
 }
 
-TEST_F(TransactionTest, TimeoutTest) {
+TEST_P(TransactionTest, TimeoutTest) {
   WriteOptions write_options;
   ReadOptions read_options;
   string value;
@@ -3243,7 +3282,7 @@ TEST_F(TransactionTest, TimeoutTest) {
   delete txn2;
 }
 
-TEST_F(TransactionTest, SingleDeleteTest) {
+TEST_P(TransactionTest, SingleDeleteTest) {
   WriteOptions write_options;
   ReadOptions read_options;
   string value;
@@ -3341,7 +3380,7 @@ TEST_F(TransactionTest, SingleDeleteTest) {
   ASSERT_TRUE(s.IsNotFound());
 }
 
-TEST_F(TransactionTest, MergeTest) {
+TEST_P(TransactionTest, MergeTest) {
   WriteOptions write_options;
   ReadOptions read_options;
   string value;
@@ -3397,7 +3436,7 @@ TEST_F(TransactionTest, MergeTest) {
   ASSERT_EQ("a,3", value);
 }
 
-TEST_F(TransactionTest, DeferSnapshotTest) {
+TEST_P(TransactionTest, DeferSnapshotTest) {
   WriteOptions write_options;
   ReadOptions read_options;
   string value;
@@ -3448,7 +3487,7 @@ TEST_F(TransactionTest, DeferSnapshotTest) {
   ASSERT_EQ("b0", value);
 }
 
-TEST_F(TransactionTest, DeferSnapshotTest2) {
+TEST_P(TransactionTest, DeferSnapshotTest2) {
   WriteOptions write_options;
   ReadOptions read_options, snapshot_read_options;
   string value;
@@ -3505,7 +3544,7 @@ TEST_F(TransactionTest, DeferSnapshotTest2) {
   delete txn1;
 }
 
-TEST_F(TransactionTest, DeferSnapshotSavePointTest) {
+TEST_P(TransactionTest, DeferSnapshotSavePointTest) {
   WriteOptions write_options;
   ReadOptions read_options, snapshot_read_options;
   string value;
@@ -3613,7 +3652,7 @@ TEST_F(TransactionTest, DeferSnapshotSavePointTest) {
   delete txn1;
 }
 
-TEST_F(TransactionTest, SetSnapshotOnNextOperationWithNotification) {
+TEST_P(TransactionTest, SetSnapshotOnNextOperationWithNotification) {
   WriteOptions write_options;
   ReadOptions read_options;
   string value;
@@ -3672,7 +3711,7 @@ TEST_F(TransactionTest, SetSnapshotOnNextOperationWithNotification) {
   delete txn1;
 }
 
-TEST_F(TransactionTest, ClearSnapshotTest) {
+TEST_P(TransactionTest, ClearSnapshotTest) {
   WriteOptions write_options;
   ReadOptions read_options, snapshot_read_options;
   string value;
@@ -3719,7 +3758,7 @@ TEST_F(TransactionTest, ClearSnapshotTest) {
   delete txn;
 }
 
-TEST_F(TransactionTest, ToggleAutoCompactionTest) {
+TEST_P(TransactionTest, ToggleAutoCompactionTest) {
   Status s;
 
   TransactionOptions txn_options;
@@ -3778,7 +3817,7 @@ TEST_F(TransactionTest, ToggleAutoCompactionTest) {
   }
 }
 
-TEST_F(TransactionTest, ExpiredTransactionDataRace1) {
+TEST_P(TransactionTest, ExpiredTransactionDataRace1) {
   // In this test, txn1 should succeed committing,
   // as the callback is called after txn1 starts committing.
   rocksdb::SyncPoint::GetInstance()->LoadDependency(
@@ -3860,7 +3899,7 @@ Status TransactionStressTestInserter(TransactionDB* db,
 }
 }  // namespace
 
-TEST_F(TransactionTest, TransactionStressTest) {
+TEST_P(TransactionTest, TransactionStressTest) {
   const size_t num_threads = 4;
   const size_t num_transactions_per_thread = 10000;
   const size_t num_sets = 3;