add flush interface to DB

Summary: as subject. The flush will flush everything in the db.

Test Plan: new test in db_test.cc

Reviewers: dhruba

Reviewed By: dhruba

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

db/db_impl.cc

@@ -548,6 +548,11 @@ int DBImpl::Level0StopWriteTrigger() {
 	return options_.level0_stop_writes_trigger;
 }
 
+Status DBImpl::Flush(const FlushOptions& options) {
+  Status status = FlushMemTable(options);
+  return status;
+}
+
 void DBImpl::TEST_CompactRange(int level, const Slice* begin,const Slice* end) {
   assert(level >= 0);
 
@@ -582,17 +587,17 @@ void DBImpl::TEST_CompactRange(int level, const Slice* begin,const Slice* end) {
   }
 }
 
-Status DBImpl::TEST_CompactMemTable() {
+Status DBImpl::FlushMemTable(const FlushOptions& options) {
   // NULL batch means just wait for earlier writes to be done
   Status s = Write(WriteOptions(), NULL);
-  if (s.ok()) {
+  if (s.ok() && options.wait) {
     // Wait until the compaction completes
-	s = TEST_WaitForCompactMemTable();
+    s = WaitForCompactMemTable();
   }
   return s;
 }
 
-Status DBImpl::TEST_WaitForCompactMemTable() {
+Status DBImpl::WaitForCompactMemTable() {
     Status s;
     // Wait until the compaction completes
     MutexLock l(&mutex_);
@@ -605,6 +610,15 @@ Status DBImpl::TEST_WaitForCompactMemTable() {
     return s;
 }
 
+
+Status DBImpl::TEST_CompactMemTable() {
+  return FlushMemTable(FlushOptions());
+}
+
+Status DBImpl::TEST_WaitForCompactMemTable() {
+	return WaitForCompactMemTable();
+}
+
 Status DBImpl::TEST_WaitForCompact() {
 	// Wait until the compaction completes
 	MutexLock l(&mutex_);

db/db_impl.h

@@ -43,6 +43,7 @@ class DBImpl : public DB {
   virtual int NumberLevels();
   virtual int MaxMemCompactionLevel();
   virtual int Level0StopWriteTrigger();
+  virtual Status Flush(const FlushOptions& options);
 
   // Extra methods (for testing) that are not in the public DB interface
 
@@ -100,6 +101,12 @@ class DBImpl : public DB {
   Status MakeRoomForWrite(bool force /* compact even if there is room? */);
   WriteBatch* BuildBatchGroup(Writer** last_writer);
 
+  // Force current memtable contents to be flushed.
+  Status FlushMemTable(const FlushOptions& options);
+
+  // Wait for memtable compaction
+  Status WaitForCompactMemTable();
+
   void MaybeScheduleCompaction();
   static void BGWork(void* db);
   void BackgroundCall();

db/db_test.cc

@@ -784,11 +784,11 @@ TEST(DBTest, WAL) {
   WriteOptions writeOpt = WriteOptions();
   writeOpt.disableWAL = true;
   ASSERT_OK(dbfull()->Put(writeOpt, "foo", "v1"));
-  ASSERT_OK(dbfull()->Put(writeOpt, "baz", "v1"));
+  ASSERT_OK(dbfull()->Put(writeOpt, "bar", "v1"));
 
   Reopen();
   ASSERT_EQ("NOT_FOUND", Get("foo"));
-  ASSERT_EQ("NOT_FOUND", Get("baz"));
+  ASSERT_EQ("NOT_FOUND", Get("bar"));
 
   writeOpt.disableWAL = false;
   ASSERT_OK(dbfull()->Put(writeOpt, "bar", "v2"));
@@ -812,6 +812,40 @@ TEST(DBTest, WAL) {
   ASSERT_EQ("v3", Get("foo"));
 }
 
+TEST(DBTest, FLUSH) {
+  Options options = CurrentOptions();
+  WriteOptions writeOpt = WriteOptions();
+  writeOpt.disableWAL = true;
+  ASSERT_OK(dbfull()->Put(writeOpt, "foo", "v1"));
+  // this will not flush the last 2 writes
+  dbfull()->Flush(FlushOptions());
+  ASSERT_OK(dbfull()->Put(writeOpt, "bar", "v1"));
+
+  Reopen();
+  ASSERT_EQ("v1", Get("foo"));
+  ASSERT_EQ("NOT_FOUND", Get("bar"));
+
+  writeOpt.disableWAL = true;
+  ASSERT_OK(dbfull()->Put(writeOpt, "bar", "v2"));
+  ASSERT_OK(dbfull()->Put(writeOpt, "foo", "v2"));
+  dbfull()->Flush(FlushOptions());
+
+  Reopen();
+  ASSERT_EQ("v2", Get("bar"));
+  ASSERT_EQ("v2", Get("foo"));
+
+  writeOpt.disableWAL = false;
+  ASSERT_OK(dbfull()->Put(writeOpt, "bar", "v3"));
+  ASSERT_OK(dbfull()->Put(writeOpt, "foo", "v3"));
+  dbfull()->Flush(FlushOptions());
+
+  Reopen();
+  // 'foo' should be there because its put
+  // has WAL enabled.
+  ASSERT_EQ("v3", Get("foo"));
+  ASSERT_EQ("v3", Get("bar"));
+}
+
 TEST(DBTest, RecoveryWithEmptyLog) {
   do {
     ASSERT_OK(Put("foo", "v1"));
@@ -1758,6 +1792,11 @@ class ModelDB: public DB {
 	return -1;
   }
 
+  virtual Status Flush(const leveldb::FlushOptions& options) {
+    Status ret;
+    return ret;
+  }
+
  private:
   class ModelIter: public Iterator {
    public:

include/leveldb/db.h

@@ -19,6 +19,7 @@ static const int kMinorVersion = 4;
 struct Options;
 struct ReadOptions;
 struct WriteOptions;
+struct FlushOptions;
 class WriteBatch;
 
 // Abstract handle to particular state of a DB.
@@ -150,6 +151,9 @@ class DB {
   // Number of files in level-0 that would stop writes.
   virtual int Level0StopWriteTrigger() = 0;
 
+  // Flush all mem-table data.
+  virtual Status Flush(const FlushOptions& options) = 0;
+
  private:
   // No copying allowed
   DB(const DB&);

include/leveldb/options.h

@@ -254,6 +254,17 @@ struct WriteOptions {
   }
 };
 
+// Options that control flush operations
+struct FlushOptions {
+  // If true, the flush will wait until the flush is done.
+  // Default: true
+  bool wait;
+
+  FlushOptions()
+      : wait(true) {
+  }
+};
+
 }  // namespace leveldb
 
 #endif  // STORAGE_LEVELDB_INCLUDE_OPTIONS_H_