Add backward compatible option in GetLiveFiles to choose whether to not Flush first

Summary:
As explained in comments in GetLiveFiles in db.h, this option will cause flush to be skipped in GetLiveFiles because some use-cases use GetSortedWalFiles after GetLiveFiles to generate more complete snapshots.
Using GetSortedWalFiles after GetLiveFiles allows us to not Flush in GetLiveFiles first because wals have everything.
Note: file deletions will be disabled before calling GLF or GSWF so live logs will not move to archive logs or get delted.
Note: Manifest file is truncated to a proper value in GLF, so it will always reply from the proper wal files on a restart

Test Plan: make

Reviewers: dhruba, haobo

Reviewed By: dhruba

CC: leveldb

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

db/db_filesnapshot.cc

@@ -30,16 +30,19 @@ Status DBImpl::EnableFileDeletions() {
 }
 
 Status DBImpl::GetLiveFiles(std::vector<std::string>& ret,
-                            uint64_t* manifest_file_size) {
+                            uint64_t* manifest_file_size,
+                            bool flush_memtable) {
 
   *manifest_file_size = 0;
 
-  // flush all dirty data to disk.
-  Status status =  Flush(FlushOptions());
-  if (!status.ok()) {
-    Log(options_.info_log, "Cannot Flush data %s\n",
-        status.ToString().c_str());
-    return status;
+  if (flush_memtable) {
+    // flush all dirty data to disk.
+    Status status =  Flush(FlushOptions());
+    if (!status.ok()) {
+      Log(options_.info_log, "Cannot Flush data %s\n",
+          status.ToString().c_str());
+      return status;
+    }
   }
 
   MutexLock l(&mutex_);

db/db_impl.h

@@ -72,7 +72,8 @@ class DBImpl : public DB {
   virtual Status DisableFileDeletions();
   virtual Status EnableFileDeletions();
   virtual Status GetLiveFiles(std::vector<std::string>&,
-                              uint64_t* manifest_file_size);
+                              uint64_t* manifest_file_size,
+                              bool flush_memtable = true);
   virtual Status GetSortedWalFiles(VectorLogPtr& files);
   virtual Status DeleteWalFiles(const VectorLogPtr& files);
   virtual SequenceNumber GetLatestSequenceNumber();

db/db_impl_readonly.h

@@ -60,7 +60,8 @@ public:
    return Status::NotSupported("Not supported operation in read only mode.");
  }
  virtual Status GetLiveFiles(std::vector<std::string>&,
-                             uint64_t* manifest_file_size) {
+                             uint64_t* manifest_file_size,
+                             bool flush_memtable = true) {
    return Status::NotSupported("Not supported operation in read only mode.");
  }
  virtual Status Flush(const FlushOptions& options) {

db/db_test.cc

@@ -3869,7 +3869,8 @@ class ModelDB: public DB {
   virtual Status EnableFileDeletions() {
     return Status::OK();
   }
-  virtual Status GetLiveFiles(std::vector<std::string>&, uint64_t* size) {
+  virtual Status GetLiveFiles(std::vector<std::string>&, uint64_t* size,
+                              bool flush_memtable = true) {
     return Status::OK();
   }
 

include/rocksdb/db.h

@@ -244,8 +244,15 @@ class DB {
   // manifest file is returned in manifest_file_size. The manifest file is an
   // ever growing file, but only the portion specified by manifest_file_size is
   // valid for this snapshot.
+  // Setting flush_memtable to true does Flush before recording the live files.
+  // Setting flush_memtable to false is useful when we don't want to wait for
+  // flush which may have to wait for compaction to complete taking an
+  // indeterminate time. But this will have to use GetSortedWalFiles after
+  // GetLiveFiles to compensate for memtables missed in this snapshot due to the
+  // absence of Flush, by WAL files to recover the database consistently later
   virtual Status GetLiveFiles(std::vector<std::string>&,
-                              uint64_t* manifest_file_size) = 0;
+                              uint64_t* manifest_file_size,
+                              bool flush_memtable = true) = 0;
 
   // Retrieve the sorted list of all wal files with earliest file first
   virtual Status GetSortedWalFiles(VectorLogPtr& files) = 0;

include/utilities/stackable_db.h

@@ -133,9 +133,9 @@ class StackableDB : public DB {
     return sdb_->EnableFileDeletions();
   }
 
-  virtual Status GetLiveFiles(std::vector<std::string>& vec, uint64_t* mfs)
-    override {
-      return sdb_->GetLiveFiles(vec, mfs);
+  virtual Status GetLiveFiles(std::vector<std::string>& vec, uint64_t* mfs,
+                              bool flush_memtable = true) override {
+      return sdb_->GetLiveFiles(vec, mfs, flush_memtable);
   }
 
   virtual SequenceNumber GetLatestSequenceNumber() override {

utilities/ttl/db_ttl.cc

@@ -267,8 +267,9 @@ Status DBWithTTL::EnableFileDeletions() {
   return db_->EnableFileDeletions();
 }
 
-Status DBWithTTL::GetLiveFiles(std::vector<std::string>& vec, uint64_t* mfs) {
-  return db_->GetLiveFiles(vec, mfs);
+Status DBWithTTL::GetLiveFiles(std::vector<std::string>& vec, uint64_t* mfs,
+                               bool flush_memtable) {
+  return db_->GetLiveFiles(vec, mfs, flush_memtable);
 }
 
 SequenceNumber DBWithTTL::GetLatestSequenceNumber() {

utilities/ttl/db_ttl.h

@@ -75,7 +75,8 @@ class DBWithTTL : public StackableDB {
 
   virtual Status EnableFileDeletions();
 
-  virtual Status GetLiveFiles(std::vector<std::string>& vec, uint64_t* mfs);
+  virtual Status GetLiveFiles(std::vector<std::string>& vec, uint64_t* mfs,
+                              bool flush_memtable = true);
 
   virtual Status GetSortedWalFiles(VectorLogPtr& files);