Add DestroyColumnFamilyHandle(ColumnFamilyHandle**) to db.h

Summary:
add DestroyColumnFamilyHandle(ColumnFamilyHandle**) to close column family instead of deleting cfh*
User should call this to close a cf and then we can detect the deletion in this function.

Test Plan: make all check -j64

Reviewers: andrewkr, yiwu, sdong

Reviewed By: sdong

Subscribers: andrewkr, dhruba, leveldb

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

db/column_family_test.cc

@@ -147,7 +147,7 @@ class ColumnFamilyTest : public testing::Test {
   void Close() {
     for (auto h : handles_) {
       if (h) {
-        delete h;
+        db_->DestroyColumnFamilyHandle(h);
       }
     }
     handles_.clear();
@@ -258,7 +258,7 @@ class ColumnFamilyTest : public testing::Test {
   void DropColumnFamilies(const std::vector<int>& cfs) {
     for (auto cf : cfs) {
       ASSERT_OK(db_->DropColumnFamily(handles_[cf]));
-      delete handles_[cf];
+      db_->DestroyColumnFamilyHandle(handles_[cf]);
       handles_[cf] = nullptr;
       names_[cf] = "";
     }
@@ -2046,7 +2046,7 @@ TEST_F(ColumnFamilyTest, ReadDroppedColumnFamily) {
         ASSERT_OK(db_->DropColumnFamily(handles_[2]));
       } else {
         // delete CF two
-        delete handles_[2];
+        db_->DestroyColumnFamilyHandle(handles_[2]);
         handles_[2] = nullptr;
       }
       // Make sure iterator created can still be used.

db/db_impl.cc

@@ -5557,6 +5557,10 @@ Status DB::CreateColumnFamily(const ColumnFamilyOptions& cf_options,
 Status DB::DropColumnFamily(ColumnFamilyHandle* column_family) {
   return Status::NotSupported("");
 }
+Status DB::DestroyColumnFamilyHandle(ColumnFamilyHandle* column_family) {
+  delete column_family;
+  return Status::OK();
+}
 
 DB::~DB() { }
 

db/db_test_util.cc

@@ -442,7 +442,7 @@ void DBTestBase::Reopen(const Options& options) {
 
 void DBTestBase::Close() {
   for (auto h : handles_) {
-    delete h;
+    db_->DestroyColumnFamilyHandle(h);
   }
   handles_.clear();
   delete db_;

include/rocksdb/db.h

@@ -147,7 +147,9 @@ class DB {
   // in rocksdb::kDefaultColumnFamilyName.
   // If everything is OK, handles will on return be the same size
   // as column_families --- handles[i] will be a handle that you
-  // will use to operate on column family column_family[i]
+  // will use to operate on column family column_family[i].
+  // Before delete DB, you have to close All column families by calling
+  // DestroyColumnFamilyHandle() with all the handles.
   static Status Open(const DBOptions& db_options, const std::string& name,
                      const std::vector<ColumnFamilyDescriptor>& column_families,
                      std::vector<ColumnFamilyHandle*>* handles, DB** dbptr);
@@ -173,6 +175,11 @@ class DB {
   // only records a drop record in the manifest and prevents the column
   // family from flushing and compacting.
   virtual Status DropColumnFamily(ColumnFamilyHandle* column_family);
+  // Close a column family specified by column_family handle and destroy
+  // the column family handle specified to avoid double deletion. This call
+  // deletes the column family handle by default. Use this method to
+  // close column family instead of deleting column family handle directly
+  virtual Status DestroyColumnFamilyHandle(ColumnFamilyHandle* column_family);
 
   // Set the database entry for "key" to "value".
   // If "key" already exists, it will be overwritten.

include/rocksdb/utilities/stackable_db.h

@@ -40,6 +40,11 @@ class StackableDB : public DB {
     return db_->DropColumnFamily(column_family);
   }
 
+  virtual Status DestroyColumnFamilyHandle(
+      ColumnFamilyHandle* column_family) override {
+    return db_->DestroyColumnFamilyHandle(column_family);
+  }
+
   using DB::Put;
   virtual Status Put(const WriteOptions& options,
                      ColumnFamilyHandle* column_family, const Slice& key,