diff --git a/Makefile b/Makefile
index ff8347957..6a5995b28 100644
--- a/Makefile
+++ b/Makefile
@@ -51,6 +51,7 @@ VALGRIND_OPTS = --error-exitcode=$(VALGRIND_ERROR) --leak-check=full
 TESTS = \
 	db_test \
 	autovector_test \
+	column_family_test \
 	table_properties_collector_test \
 	arena_test \
 	auto_roll_logger_test \
@@ -230,6 +231,9 @@ arena_test: util/arena_test.o $(LIBOBJECTS) $(TESTHARNESS)
 autovector_test: util/autovector_test.o $(LIBOBJECTS) $(TESTHARNESS)
 	$(CXX) util/autovector_test.o $(LIBOBJECTS) $(TESTHARNESS) $(EXEC_LDFLAGS) -o $@ $(LDFLAGS) $(COVERAGEFLAGS)
 
+column_family_test: db/column_family_test.o $(LIBOBJECTS) $(TESTHARNESS)
+	$(CXX) db/column_family_test.o $(LIBOBJECTS) $(TESTHARNESS) $(EXEC_LDFLAGS) -o $@ $(LDFLAGS) $(COVERAGEFLAGS)
+
 table_properties_collector_test: db/table_properties_collector_test.o $(LIBOBJECTS) $(TESTHARNESS)
 	$(CXX) db/table_properties_collector_test.o $(LIBOBJECTS) $(TESTHARNESS) $(EXEC_LDFLAGS) -o $@ $(LDFLAGS) $(COVERAGEFLAGS)
 
diff --git a/db/column_family_test.cc b/db/column_family_test.cc
new file mode 100644
index 000000000..d0da6e81f
--- /dev/null
+++ b/db/column_family_test.cc
@@ -0,0 +1,73 @@
+//  Copyright (c) 2013, Facebook, Inc.  All rights reserved.
+//  This source code is licensed under the BSD-style license found in the
+//  LICENSE file in the root directory of this source tree. An additional grant
+//  of patent rights can be found in the PATENTS file in the same directory.
+//
+// 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 "db/db_impl.h"
+#include "rocksdb/db.h"
+#include "util/testharness.h"
+
+#include <algorithm>
+#include <vector>
+#include <string>
+
+namespace rocksdb {
+
+using namespace std;
+
+class ColumnFamilyTest {
+ public:
+  ColumnFamilyTest() {
+    dbname_ = test::TmpDir() + "/column_family_test";
+    db_options_.create_if_missing = true;
+    options_.create_if_missing = true;
+    DestroyDB(dbname_, options_);
+  }
+
+  void Close() {
+    delete db_;
+    db_ = nullptr;
+  }
+
+  void Open() {
+    ASSERT_OK(DB::Open(options_, dbname_, &db_));
+  }
+
+  Options options_;
+  ColumnFamilyOptions column_family_options_;
+  DBOptions db_options_;
+  string dbname_;
+  DB* db_;
+};
+
+TEST(ColumnFamilyTest, AddDrop) {
+  Open();
+  ColumnFamilyHandle handles[4];
+  ASSERT_OK(db_->CreateColumnFamily(column_family_options_, Slice("one"),
+                                    &handles[0]));
+  ASSERT_OK(db_->CreateColumnFamily(column_family_options_, Slice("two"),
+                                    &handles[1]));
+  ASSERT_OK(db_->CreateColumnFamily(column_family_options_, Slice("three"),
+                                    &handles[2]));
+  ASSERT_OK(db_->DropColumnFamily(handles[1]));
+  ASSERT_OK(db_->CreateColumnFamily(column_family_options_, Slice("four"),
+                                    &handles[3]));
+  Close();
+  Open(); // this will roll the manifest, column families should stay consistent
+  Close();
+
+  vector<string> families;
+  DB::ListColumnFamilies(db_options_, dbname_, &families);
+  sort(families.begin(), families.end());
+  ASSERT_TRUE(families == vector<string>({"four", "one", "three"}));
+}
+
+}  // namespace rocksdb
+
+int main(int argc, char** argv) {
+  return rocksdb::test::RunAllTests();
+}
diff --git a/db/db_impl.cc b/db/db_impl.cc
index c6b37b5fb..9fe730e61 100644
--- a/db/db_impl.cc
+++ b/db/db_impl.cc
@@ -2856,6 +2856,26 @@ std::vector<Status> DBImpl::MultiGet(
   return statList;
 }
 
+Status DBImpl::CreateColumnFamily(const ColumnFamilyOptions& options,
+                                  const Slice& column_family,
+                                  ColumnFamilyHandle* handle) {
+  VersionEdit edit(0);
+  edit.AddColumnFamily(column_family.ToString());
+  MutexLock l(&mutex_);
+  ++versions_->max_column_family_;
+  handle->id = versions_->max_column_family_;
+  edit.SetColumnFamily(handle->id);
+  return versions_->LogAndApply(&edit, &mutex_);
+}
+
+Status DBImpl::DropColumnFamily(const ColumnFamilyHandle& column_family) {
+  VersionEdit edit(0);
+  edit.DropColumnFamily();
+  edit.SetColumnFamily(column_family.id);
+  MutexLock l(&mutex_);
+  return versions_->LogAndApply(&edit, &mutex_);
+}
+
 bool DBImpl::KeyMayExist(const ReadOptions& options,
                          const ColumnFamilyHandle& column_family,
                          const Slice& key, std::string* value,
@@ -3776,14 +3796,14 @@ Status DB::Merge(const WriteOptions& opt,
   return Write(opt, &batch);
 }
 
-Status DB::OpenColumnFamily(const ColumnFamilyOptions& options,
-                            const Slice& column_family,
-                            ColumnFamilyHandle* handle) {
-  return Status::NotSupported("working on it");
+// Default implementation -- returns not supported status
+Status DB::CreateColumnFamily(const ColumnFamilyOptions& options,
+                              const Slice& column_family,
+                              ColumnFamilyHandle* handle) {
+  return Status::NotSupported("");
 }
-
 Status DB::DropColumnFamily(const ColumnFamilyHandle& column_family) {
-  return Status::NotSupported("working on it");
+  return Status::NotSupported("");
 }
 
 DB::~DB() { }
@@ -3866,10 +3886,25 @@ Status DB::OpenWithColumnFamilies(
   return Status::NotSupported("Working on it");
 }
 
-Status DB::ListColumnFamilies(
-    const DBOptions& db_options, const std::string& name,
-    const std::vector<std::string>* column_families) {
-  // TODO
+Status DB::ListColumnFamilies(const DBOptions& db_options,
+                              const std::string& name,
+                              std::vector<std::string>* column_families) {
+  Options options(db_options, ColumnFamilyOptions());
+  InternalKeyComparator* icmp = new InternalKeyComparator(options.comparator);
+  TableCache* table_cache = new TableCache(name, &options, EnvOptions(options),
+                                           db_options.max_open_files - 10);
+  VersionSet* version_set =
+      new VersionSet(name, &options, EnvOptions(options), table_cache, icmp);
+
+  version_set->Recover();
+  column_families->clear();
+  for (auto cf : version_set->column_families_) {
+    column_families->push_back(cf.first);
+  }
+
+  delete version_set;
+  delete table_cache;
+  delete icmp;
   return Status::NotSupported("Working on it");
 }
 
diff --git a/db/db_impl.h b/db/db_impl.h
index 18ed18df7..de8ef288b 100644
--- a/db/db_impl.h
+++ b/db/db_impl.h
@@ -61,6 +61,11 @@ class DBImpl : public DB {
       const std::vector<ColumnFamilyHandle>& column_family,
       const std::vector<Slice>& keys, std::vector<std::string>* values);
 
+  virtual Status CreateColumnFamily(const ColumnFamilyOptions& options,
+                                    const Slice& column_family,
+                                    ColumnFamilyHandle* handle);
+  virtual Status DropColumnFamily(const ColumnFamilyHandle& column_family);
+
   // Returns false if key doesn't exist in the database and true if it may.
   // If value_found is not passed in as null, then return the value if found in
   // memory. On return, if value was found, then value_found will be set to true
diff --git a/db/db_impl_readonly.cc b/db/db_impl_readonly.cc
index 3e906246e..fe9c8e2bd 100644
--- a/db/db_impl_readonly.cc
+++ b/db/db_impl_readonly.cc
@@ -52,8 +52,9 @@ DBImplReadOnly::~DBImplReadOnly() {
 }
 
 // Implementations of the DB interface
-Status DBImplReadOnly::Get(const ReadOptions& options, const Slice& key,
-                           std::string* value) {
+Status DBImplReadOnly::Get(const ReadOptions& options,
+                           const ColumnFamilyHandle& column_family,
+                           const Slice& key, std::string* value) {
   Status s;
   MemTable* mem = GetMemTable();
   Version* current = versions_->current();
diff --git a/db/db_impl_readonly.h b/db/db_impl_readonly.h
index 0169f57f6..8bfaefaae 100644
--- a/db/db_impl_readonly.h
+++ b/db/db_impl_readonly.h
@@ -28,9 +28,9 @@ public:
  virtual ~DBImplReadOnly();
 
  // Implementations of the DB interface
- using DBImpl::Get;
+ using DB::Get;
  virtual Status Get(const ReadOptions& options,
-                    const Slice& key,
+                    const ColumnFamilyHandle& column_family, const Slice& key,
                     std::string* value);
 
  // TODO: Implement ReadOnly MultiGet?
diff --git a/db/version_set.cc b/db/version_set.cc
index d48f9269d..1bba25dd4 100644
--- a/db/version_set.cc
+++ b/db/version_set.cc
@@ -1466,6 +1466,23 @@ Status VersionSet::Recover() {
         builder.Apply(&edit);
       }
 
+      if (edit.is_column_family_add_) {
+        assert(column_families_.find(edit.column_family_name_) ==
+               column_families_.end());
+        column_families_.insert(
+            {edit.column_family_name_, edit.column_family_});
+        column_family_data_.insert(
+            {edit.column_family_, ColumnFamilyData(edit.column_family_name_)});
+        max_column_family_ = std::max(max_column_family_, edit.column_family_);
+      }
+
+      if (edit.is_column_family_drop_) {
+        auto cf = column_family_data_.find(edit.column_family_);
+        assert(cf != column_family_data_.end());
+        column_families_.erase(cf->second.name);
+        column_family_data_.erase(cf);
+      }
+
       if (edit.has_log_number_) {
         log_number = edit.log_number_;
         have_log_number = true;
@@ -1827,6 +1844,19 @@ void VersionSet::UpdateFilesBySize(Version* v) {
 Status VersionSet::WriteSnapshot(log::Writer* log) {
   // TODO: Break up into multiple records to reduce memory usage on recovery?
 
+  // Save column families
+  for (auto cf : column_families_) {
+    VersionEdit edit(0);
+    edit.AddColumnFamily(cf.first);
+    edit.SetColumnFamily(cf.second);
+    std::string record;
+    edit.EncodeTo(&record);
+    Status s = log->AddRecord(record);
+    if (!s.ok()) {
+      return s;
+    }
+  }
+
   // Save metadata
   VersionEdit edit(NumberLevels());
   edit.SetComparatorName(icmp_.user_comparator()->Name());
diff --git a/db/version_set.h b/db/version_set.h
index 1b62e5193..8d26129eb 100644
--- a/db/version_set.h
+++ b/db/version_set.h
@@ -436,6 +436,15 @@ class VersionSet {
 
   void GetObsoleteFiles(std::vector<FileMetaData*>* files);
 
+  // column family metadata
+  struct ColumnFamilyData {
+    std::string name;
+    explicit ColumnFamilyData(const std::string& name) : name(name) {}
+  };
+  std::unordered_map<std::string, uint32_t> column_families_;
+  std::unordered_map<uint32_t, ColumnFamilyData> column_family_data_;
+  uint32_t max_column_family_;
+
  private:
   class Builder;
   struct ManifestWriter;
@@ -505,10 +514,6 @@ class VersionSet {
   // generates a increasing version number for every new version
   uint64_t current_version_number_;
 
-  // column family metadata
-  std::unordered_map<std::string, ColumnFamilyHandle> column_families_;
-  uint32_t max_column_family_id_;
-
   // Queue of writers to the manifest file
   std::deque<ManifestWriter*> manifest_writers_;
 
diff --git a/include/rocksdb/db.h b/include/rocksdb/db.h
index 03339784e..bb2af14b0 100644
--- a/include/rocksdb/db.h
+++ b/include/rocksdb/db.h
@@ -107,28 +107,24 @@ class DB {
   // and return the list of all column families in that DB
   // through column_families argument. The ordering of
   // column families in column_families is unspecified.
-  static Status ListColumnFamilies(
-      const DBOptions& db_options, const std::string& name,
-      const std::vector<std::string>* column_families);
+  static Status ListColumnFamilies(const DBOptions& db_options,
+                                   const std::string& name,
+                                   std::vector<std::string>* column_families);
 
   DB() { }
   virtual ~DB();
 
-  // Open a column_family and return the handle of column family
-  // through the argument handle
-  // If the column family already exists in the Database,
-  // it will open it and make it available for the client to query.
-  // If the column family does not exist, the function will create
-  // and persist it.
-  Status OpenColumnFamily(const ColumnFamilyOptions& options,
-                          const Slice& column_family,
-                          ColumnFamilyHandle* handle);
+  // Create a column_family and return the handle of column family
+  // through the argument handle.
+  virtual Status CreateColumnFamily(const ColumnFamilyOptions& options,
+                                    const Slice& column_family,
+                                    ColumnFamilyHandle* handle);
 
   // Drop a column family specified by column_family handle.
   // All data related to the column family will be deleted before
   // the function returns.
   // Calls referring to the dropped column family will fail.
-  Status DropColumnFamily(const ColumnFamilyHandle& column_family);
+  virtual Status DropColumnFamily(const ColumnFamilyHandle& column_family);
 
   // Set the database entry for "key" to "value".
   // Returns OK on success, and a non-OK status on error.