Enhancements to rocksdb for better support for replication.

Summary:
1. The OpenForReadOnly() call should not lock the db. This is useful
so that multiple processes can open the same database concurrently
for reading.
2. GetUpdatesSince should not error out if the archive directory
does not exist.
3. A new constructor for WriteBatch that can takes a serialized
string as a parameter of the constructor.

Test Plan: make clean check

Reviewers: sheki

Reviewed By: sheki

CC: leveldb

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

.arcconfig

@@ -1,10 +1,5 @@
 {
   "project_id" : "leveldb",
   "conduit_uri" : "https://reviews.facebook.net/",
-  "copyright_holder" : "",
-  "load" : [
-    "linters/src/"
-  ],
-  "lint_engine" : "FacebookFbcodeLintEngine", 
-  "lint.engine.single.linter" : "FbcodeCppLinter" 
+  "copyright_holder" : ""
 }

db/db_impl.cc

@@ -512,31 +512,33 @@ Status DBImpl::Recover(VersionEdit* edit, bool no_log_recory,
   // 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.
-  env_->CreateDir(dbname_);
   assert(db_lock_ == NULL);
-  Status s = env_->LockFile(LockFileName(dbname_), &db_lock_);
-  if (!s.ok()) {
-    return s;
-  }
+  if (!no_log_recory) {
+    env_->CreateDir(dbname_);
+    Status s = env_->LockFile(LockFileName(dbname_), &db_lock_);
+    if (!s.ok()) {
+      return s;
+    }
 
-  if (!env_->FileExists(CurrentFileName(dbname_))) {
-    if (options_.create_if_missing) {
-      s = NewDB();
-      if (!s.ok()) {
-        return s;
+    if (!env_->FileExists(CurrentFileName(dbname_))) {
+      if (options_.create_if_missing) {
+        s = NewDB();
+        if (!s.ok()) {
+          return s;
+        }
+      } else {
+        return Status::InvalidArgument(
+            dbname_, "does not exist (create_if_missing is false)");
       }
     } else {
-      return Status::InvalidArgument(
-          dbname_, "does not exist (create_if_missing is false)");
-    }
-  } else {
-    if (options_.error_if_exists) {
-      return Status::InvalidArgument(
-          dbname_, "exists (error_if_exists is true)");
+      if (options_.error_if_exists) {
+        return Status::InvalidArgument(
+            dbname_, "exists (error_if_exists is true)");
+      }
     }
   }
 
-  s = versions_->Recover();
+  Status s = versions_->Recover();
   if (s.ok()) {
     SequenceNumber max_sequence(0);
 
@@ -897,9 +899,12 @@ Status DBImpl::GetUpdatesSince(SequenceNumber seq,
     return s;
   }
   //  list wal files in archive dir.
-  s = ListAllWALFiles(ArchivalDirectory(dbname_), &walFiles, kArchivedLogFile);
-  if (!s.ok()) {
-    return s;
+  std::string archivedir = ArchivalDirectory(dbname_);
+  if (env_->FileExists(archivedir)) {
+    s = ListAllWALFiles(archivedir, &walFiles, kArchivedLogFile);
+    if (!s.ok()) {
+      return s;
+    }
   }
 
   if (walFiles.empty()) {

db/transaction_log_iterator_impl.cc

@@ -16,8 +16,8 @@ TransactionLogIteratorImpl::TransactionLogIteratorImpl(
   isValid_(true),
   currentFileIndex_(0),
   currentLogReader_(NULL) {
-    assert( files_ != NULL);
-  }
+  assert( files_ != NULL);
+}
 
 LogReporter
 TransactionLogIteratorImpl::NewLogReporter(const uint64_t logNumber) {

include/leveldb/write_batch.h

@@ -51,9 +51,12 @@ class WriteBatch {
   };
   Status Iterate(Handler* handler) const;
 
-  // Returns the serialized string
+  // Retrive the serialized version of this batch.
   std::string Data() { return rep_; }
 
+  // Constructor with a serialized string object
+  WriteBatch(std::string rep): rep_(rep) {}
+
  private:
   friend class WriteBatchInternal;