Added clearer error message for failure to create db directory in DBImpl::Recover()

Summary:
Changed CreateDir() to CreateDirIfMissing() so a directory that already exists now causes and error.

Fixed CreateDirIfMissing() and added Env.DirExists()

Test Plan:
make check to test for regessions

Ran the following to test if the error message is not about lock files not existing
./db_bench --db=dir/testdb

After creating a file "testdb", ran the following to see if it failed with sane error message:
./db_bench --db=testdb

Reviewers: dhruba, emayanke, vamsi, sheki

Reviewed By: emayanke

CC: leveldb

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

db/db_impl.cc

@@ -511,13 +511,21 @@ Status DBImpl::Recover(VersionEdit* edit, MemTable* external_table,
     bool error_if_log_file_exist) {
   mutex_.AssertHeld();
 
-  // Ignore error from CreateDir since the creation of the DB is
-  // committed only when the descriptor is created, and this directory
-  // may already exist from a previous failed creation attempt.
   assert(db_lock_ == NULL);
   if (!external_table) {
-    env_->CreateDir(dbname_);
-    Status s = env_->LockFile(LockFileName(dbname_), &db_lock_);
+    // We call CreateDirIfMissing() as the directory may already exist (if we
+    // are reopening a DB), when this happens we don't want creating the
+    // directory to cause an error. However, we need to check if creating the
+    // directory fails or else we may get an obscure message about the lock
+    // file not existing. One real-world example of this occurring is if
+    // env->CreateDirIfMissing() doesn't create intermediate directories, e.g.
+    // when dbname_ is "dir/db" but when "dir" doesn't exist.
+    Status s = env_->CreateDirIfMissing(dbname_);
+    if (!s.ok()) {
+      return s;
+    }
+
+    s = env_->LockFile(LockFileName(dbname_), &db_lock_);
     if (!s.ok()) {
       return s;
     }

hdfs/env_hdfs.h

@@ -162,7 +162,7 @@ class HdfsEnv : public Env {
                         // posix threads, etc.
 
   /**
-   * If the URI is specified of the form hdfs://server:port/path, 
+   * If the URI is specified of the form hdfs://server:port/path,
    * then connect to the specified cluster
    * else connect to default.
    */
@@ -189,7 +189,7 @@ class HdfsEnv : public Env {
     int rem = remaining.find(pathsep);
     std::string portStr = (rem == 0 ? remaining :
                            remaining.substr(0, rem));
-    
+
     tPort port;
     port = atoi(portStr.c_str());
     if (port == 0) {
@@ -199,7 +199,7 @@ class HdfsEnv : public Env {
     return fs;
   }
 
-  void split(const std::string &s, char delim, 
+  void split(const std::string &s, char delim,
              std::vector<std::string> &elems) {
     elems.clear();
     size_t prev = 0;

util/env_posix.cc

@@ -606,6 +606,10 @@ class PosixEnv : public Env {
     if (mkdir(name.c_str(), 0755) != 0) {
       if (errno != EEXIST) {
         result = IOError(name, errno);
+      } else if (!DirExists(name)) { // Check that name is actually a
+                                     // directory.
+        // Message is taken from mkdir
+        result = Status::IOError("`"+name+"' exists but is not a directory");
       }
     }
     return result;
@@ -796,6 +800,16 @@ class PosixEnv : public Env {
     }
   }
 
+  // Returns true iff the named directory exists and is a directory.
+  virtual bool DirExists(const std::string& dname) {
+    struct stat statbuf;
+    if (stat(dname.c_str(), &statbuf) == 0) {
+      return S_ISDIR(statbuf.st_mode);
+    }
+    return false; // stat() failed return false
+  }
+
+
   // BGThread() is the body of the background thread
   void BGThread();
   static void* BGThreadWrapper(void* arg) {