From 25c8a1a20f88317e7fcd0804aeebadec0214abae Mon Sep 17 00:00:00 2001
From: Igor Canadi <icanadi@fb.com>
Date: Wed, 12 Mar 2014 11:50:10 -0700
Subject: [PATCH] More bug fixed introduced by code cleanup

---
 db/db_impl.cc                              | 23 ++++++++++-----
 db/db_test.cc                              | 33 +++++++++++-----------
 utilities/backupable/backupable_db_test.cc |  2 ++
 3 files changed, 35 insertions(+), 23 deletions(-)

diff --git a/db/db_impl.cc b/db/db_impl.cc
index 759da4a996..1178d0ea4b 100644
--- a/db/db_impl.cc
+++ b/db/db_impl.cc
@@ -277,16 +277,25 @@ DBImpl::DBImpl(const DBOptions& options, const std::string& dbname)
 }
 
 DBImpl::~DBImpl() {
-  // only the default CFD is alive at this point
-  if (default_cf_handle_ != nullptr) {
-    auto default_cfd = default_cf_handle_->cfd();
-    if (flush_on_destroy_ &&
-        default_cfd->mem()->GetFirstSequenceNumber() != 0) {
-      FlushMemTable(default_cfd, FlushOptions());
+  mutex_.Lock();
+  if (flush_on_destroy_) {
+    autovector<ColumnFamilyData*> to_delete;
+    for (auto cfd : *versions_->GetColumnFamilySet()) {
+      if (cfd->mem()->GetFirstSequenceNumber() != 0) {
+        cfd->Ref();
+        mutex_.Unlock();
+        FlushMemTable(cfd, FlushOptions());
+        mutex_.Lock();
+        if (cfd->Unref()) {
+          to_delete.push_back(cfd);
+        }
+      }
+    }
+    for (auto cfd : to_delete) {
+      delete cfd;
     }
   }
 
-  mutex_.Lock();
   // Wait for background work to finish
   shutting_down_.Release_Store(this);  // Any non-nullptr value is ok
   while (bg_compaction_scheduled_ ||
diff --git a/db/db_test.cc b/db/db_test.cc
index 466f926f01..5b97fa3838 100644
--- a/db/db_test.cc
+++ b/db/db_test.cc
@@ -1998,34 +1998,35 @@ TEST(DBTest, RollLog) {
 
 TEST(DBTest, WAL) {
   do {
+    CreateAndReopenWithCF({"pikachu"});
     WriteOptions writeOpt = WriteOptions();
     writeOpt.disableWAL = true;
-    ASSERT_OK(dbfull()->Put(writeOpt, "foo", "v1"));
-    ASSERT_OK(dbfull()->Put(writeOpt, "bar", "v1"));
+    ASSERT_OK(dbfull()->Put(writeOpt, handles_[1], "foo", "v1"));
+    ASSERT_OK(dbfull()->Put(writeOpt, handles_[1], "bar", "v1"));
 
-    Reopen();
-    ASSERT_EQ("v1", Get("foo"));
-    ASSERT_EQ("v1", Get("bar"));
+    ReopenWithColumnFamilies({"default", "pikachu"});
+    ASSERT_EQ("v1", Get(1, "foo"));
+    ASSERT_EQ("v1", Get(1, "bar"));
 
     writeOpt.disableWAL = false;
-    ASSERT_OK(dbfull()->Put(writeOpt, "bar", "v2"));
+    ASSERT_OK(dbfull()->Put(writeOpt, handles_[1], "bar", "v2"));
     writeOpt.disableWAL = true;
-    ASSERT_OK(dbfull()->Put(writeOpt, "foo", "v2"));
+    ASSERT_OK(dbfull()->Put(writeOpt, handles_[1], "foo", "v2"));
 
-    Reopen();
+    ReopenWithColumnFamilies({"default", "pikachu"});
     // Both value's should be present.
-    ASSERT_EQ("v2", Get("bar"));
-    ASSERT_EQ("v2", Get("foo"));
+    ASSERT_EQ("v2", Get(1, "bar"));
+    ASSERT_EQ("v2", Get(1, "foo"));
 
     writeOpt.disableWAL = true;
-    ASSERT_OK(dbfull()->Put(writeOpt, "bar", "v3"));
+    ASSERT_OK(dbfull()->Put(writeOpt, handles_[1], "bar", "v3"));
     writeOpt.disableWAL = false;
-    ASSERT_OK(dbfull()->Put(writeOpt, "foo", "v3"));
+    ASSERT_OK(dbfull()->Put(writeOpt, handles_[1], "foo", "v3"));
 
-    Reopen();
+    ReopenWithColumnFamilies({"default", "pikachu"});
     // again both values should be present.
-    ASSERT_EQ("v3", Get("foo"));
-    ASSERT_EQ("v3", Get("bar"));
+    ASSERT_EQ("v3", Get(1, "foo"));
+    ASSERT_EQ("v3", Get(1, "bar"));
   } while (ChangeCompactOptions());
 }
 
@@ -2130,7 +2131,7 @@ TEST(DBTest, FLUSH) {
 
     ReopenWithColumnFamilies({"default", "pikachu"});
     ASSERT_EQ("v1", Get(1, "foo"));
-    ASSERT_EQ("NOT_FOUND", Get(1, "bar"));
+    ASSERT_EQ("v1", Get(1, "bar"));
 
     writeOpt.disableWAL = true;
     ASSERT_OK(dbfull()->Put(writeOpt, handles_[1], "bar", "v2"));
diff --git a/utilities/backupable/backupable_db_test.cc b/utilities/backupable/backupable_db_test.cc
index 5c20579b44..74e703e9eb 100644
--- a/utilities/backupable/backupable_db_test.cc
+++ b/utilities/backupable/backupable_db_test.cc
@@ -721,6 +721,8 @@ TEST(BackupableDBTest, FailOverwritingBackups) {
   OpenBackupableDB(true);
   for (int i = 0; i < 5; ++i) {
     FillDB(db_.get(), 100 * i, 100 * (i + 1));
+    CloseBackupableDB();
+    OpenBackupableDB(false);
     ASSERT_OK(db_->CreateNewBackup(true));
     CloseBackupableDB();
     OpenBackupableDB(false);