When creating a new DB, fail it when wal_dir contains existing log files

Summary: Current behavior of creating new DB is, if there is existing log files, we will go ahead and replay them on top of empty DB. This is a behavior that no user would expect. With this patch, we will fail the creation if a user creates a DB with existing log files.

Test Plan: make all check

Reviewers: haobo, igor, ljin

Reviewed By: haobo

CC: nkg-, yhchiang, dhruba, leveldb

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

db/db_impl.cc

@@ -1062,6 +1062,7 @@ Status DBImpl::Recover(
     bool error_if_log_file_exist) {
   mutex_.AssertHeld();
 
+  bool is_new_db = false;
   assert(db_lock_ == nullptr);
   if (!read_only) {
     // We call CreateDirIfMissing() as the directory may already exist (if we
@@ -1090,6 +1091,7 @@ Status DBImpl::Recover(
       if (options_.create_if_missing) {
         // TODO: add merge_operator name check
         s = NewDB();
+        is_new_db = true;
         if (!s.ok()) {
           return s;
         }
@@ -1140,12 +1142,17 @@ Status DBImpl::Recover(
     for (size_t i = 0; i < filenames.size(); i++) {
       uint64_t number;
       FileType type;
-      if (ParseFileName(filenames[i], &number, &type)
-          && type == kLogFile
-          && ((number >= min_log) || (number == prev_log))) {
+      if (ParseFileName(filenames[i], &number, &type) && type == kLogFile) {
+        if (is_new_db) {
+          return Status::Corruption(
+              "While creating a new Db, wal_dir contains "
+              "existing log file: ",
+              filenames[i]);
+        } else if ((number >= min_log) || (number == prev_log)) {
           logs.push_back(number);
         }
       }
+    }
 
     if (logs.size() > 0 && error_if_log_file_exist) {
       return Status::Corruption(""

db/db_test.cc

@@ -2095,14 +2095,14 @@ TEST(DBTest, IgnoreRecoveredLog) {
     ASSERT_EQ(one, Get("bar"));
     Close();
     Destroy(&options);
+    Reopen(&options);
+    Close();
 
     // copy the logs from backup back to wal dir
     env_->CreateDirIfMissing(options.wal_dir);
     for (auto& log : logs) {
       if (log != ".." && log != ".") {
         CopyFile(backup_logs + "/" + log, options.wal_dir + "/" + log);
-        // we won't be needing this file no more
-        env_->DeleteFile(backup_logs + "/" + log);
       }
     }
     // assert that we successfully recovered only from logs, even though we
@@ -2110,7 +2110,20 @@ TEST(DBTest, IgnoreRecoveredLog) {
     Reopen(&options);
     ASSERT_EQ(two, Get("foo"));
     ASSERT_EQ(one, Get("bar"));
-    Close();
+
+    // Recovery will fail if DB directory doesn't exist.
+    Destroy(&options);
+    // copy the logs from backup back to wal dir
+    env_->CreateDirIfMissing(options.wal_dir);
+    for (auto& log : logs) {
+      if (log != ".." && log != ".") {
+        CopyFile(backup_logs + "/" + log, options.wal_dir + "/" + log);
+        // we won't be needing this file no more
+        env_->DeleteFile(backup_logs + "/" + log);
+      }
+    }
+    Status s = TryReopen(&options);
+    ASSERT_TRUE(!s.ok());
   } while (ChangeOptions());
 }