add disable WAL option

Summary: add disable WAL option

Test Plan: new testcase in db_test.cc

Reviewers: dhruba

Reviewed By: dhruba

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

db/db_impl.cc

@@ -40,6 +40,7 @@ struct DBImpl::Writer {
   Status status;
   WriteBatch* batch;
   bool sync;
+  bool disableWAL;
   bool done;
   port::CondVar cv;
 
@@ -1140,6 +1141,7 @@ Status DBImpl::Write(const WriteOptions& options, WriteBatch* my_batch) {
   Writer w(&mutex_);
   w.batch = my_batch;
   w.sync = options.sync;
+  w.disableWAL = options.disableWAL;
   w.done = false;
 
   MutexLock l(&mutex_);
@@ -1166,9 +1168,11 @@ Status DBImpl::Write(const WriteOptions& options, WriteBatch* my_batch) {
     // into mem_.
     {
       mutex_.Unlock();
-      status = log_->AddRecord(WriteBatchInternal::Contents(updates));
-      if (status.ok() && options.sync) {
-        status = logfile_->Sync();
+      if (!options.disableWAL) {
+        status = log_->AddRecord(WriteBatchInternal::Contents(updates));
+        if (status.ok() && options.sync) {
+          status = logfile_->Sync();
+        }
       }
       if (status.ok()) {
         status = WriteBatchInternal::InsertInto(updates, mem_);
@@ -1227,6 +1231,12 @@ WriteBatch* DBImpl::BuildBatchGroup(Writer** last_writer) {
       break;
     }
 
+    if (!w->disableWAL && first->disableWAL) {
+      // Do not include a write that needs WAL into a batch that has
+      // WAL disabled.
+      break;
+    }
+
     if (w->batch != NULL) {
       size += WriteBatchInternal::ByteSize(w->batch);
       if (size > max_size) {

db/db_test.cc

@@ -779,6 +779,39 @@ TEST(DBTest, Recover) {
   } while (ChangeOptions());
 }
 
+TEST(DBTest, WAL) {
+  Options options = CurrentOptions();
+  WriteOptions writeOpt = WriteOptions();
+  writeOpt.disableWAL = true;
+  ASSERT_OK(dbfull()->Put(writeOpt, "foo", "v1"));
+  ASSERT_OK(dbfull()->Put(writeOpt, "baz", "v1"));
+
+  Reopen();
+  ASSERT_EQ("NOT_FOUND", Get("foo"));
+  ASSERT_EQ("NOT_FOUND", Get("baz"));
+
+  writeOpt.disableWAL = false;
+  ASSERT_OK(dbfull()->Put(writeOpt, "bar", "v2"));
+  writeOpt.disableWAL = true;
+  ASSERT_OK(dbfull()->Put(writeOpt, "foo", "v2"));
+
+  Reopen();
+  // We garantee the 'bar' will be there
+  // because its put has WAL enabled.
+  // But 'foo' may or may not be there.
+  ASSERT_EQ("v2", Get("bar"));
+
+  writeOpt.disableWAL = true;
+  ASSERT_OK(dbfull()->Put(writeOpt, "bar", "v3"));
+  writeOpt.disableWAL = false;
+  ASSERT_OK(dbfull()->Put(writeOpt, "foo", "v3"));
+
+  Reopen();
+  // 'foo' should be there because its put
+  // has WAL enabled.
+  ASSERT_EQ("v3", Get("foo"));
+}
+
 TEST(DBTest, RecoveryWithEmptyLog) {
   do {
     ASSERT_OK(Put("foo", "v1"));

include/leveldb/options.h

@@ -244,8 +244,13 @@ struct WriteOptions {
   // Default: false
   bool sync;
 
+  // If true, writes will not first go to the write ahead log,
+  // and the write may got lost after a crash.
+  bool disableWAL;
+
   WriteOptions()
-      : sync(false) {
+      : sync(false),
+        disableWAL(false) {
   }
 };