Ignore a zero-sized file while looking for a seq-no in GetUpdatesSince

Summary:
Rocksdb can create 0 sized log files when it is opened and closed without any operations.
The GetUpdatesSince fails currently if there is a log file of size zero.

This diff fixes this. If there is a log file is 0, it is removed form the probable_file_list

Test Plan: unit test

Reviewers: dhruba, heyongqiang

Reviewed By: heyongqiang

CC: leveldb

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

db/db_impl.cc

@@ -927,6 +927,13 @@ Status DBImpl::FindProbableWALFiles(std::vector<LogFile>* const allLogs,
     WriteBatch batch;
     Status s = ReadFirstRecord(allLogs->at(mid), &batch);
     if (!s.ok()) {
+      if (CheckFileExistsAndEmpty(allLogs->at(mid))) {
+        allLogs->erase(allLogs->begin() + mid);
+        if (mid == start) {
+          ++start;
+        }
+        continue;
+      }
       return s;
     }
     SequenceNumber currentSeqNum = WriteBatchInternal::Sequence(&batch);
@@ -947,6 +954,24 @@ Status DBImpl::FindProbableWALFiles(std::vector<LogFile>* const allLogs,
   return Status::OK();
 }
 
+bool DBImpl::CheckFileExistsAndEmpty(const LogFile& file) {
+  if (file.type == kAliveLogFile) {
+    const std::string fname = LogFileName(dbname_, file.logNumber);
+    uint64_t file_size;
+    Status s = env_->GetFileSize(fname, &file_size);
+    if (s.ok() && file_size == 0) {
+      return true;
+    }
+  }
+  const std::string fname = ArchivedLogFileName(dbname_, file.logNumber);
+  uint64_t file_size;
+  Status s = env_->GetFileSize(fname, &file_size);
+  if (s.ok() && file_size == 0) {
+    return true;
+  }
+  return false;
+}
+
 Status DBImpl::ReadFirstRecord(const LogFile& file, WriteBatch* const result) {
 
   if (file.type == kAliveLogFile) {

db/db_impl.h

@@ -194,6 +194,8 @@ class DBImpl : public DB {
   Status FindProbableWALFiles(std::vector<LogFile>* const allLogs,
                               std::vector<LogFile>* const result,
                               const SequenceNumber target);
+  //  return true if
+  bool CheckFileExistsAndEmpty(const LogFile& file);
 
 
   Status ReadFirstRecord(const LogFile& file, WriteBatch* const result);

db/db_test.cc

@@ -2638,6 +2638,38 @@ TEST(DBTest, TransactionLogIterator) {
   }
 }
 
+TEST(DBTest, TransactionLogIteratorMoveOverZeroFiles) {
+  std::string value(1024, '1');
+  Options options = CurrentOptions();
+  options.create_if_missing = true;
+  options.WAL_ttl_seconds = 1000;
+  DestroyAndReopen(&options);
+  // Do a plain Reopen.
+  Put("key1", value);
+  // Two reopens should create a zero record WAL file.
+  Reopen(&options);
+  Reopen(&options);
+
+  Put("key2", value);
+  unique_ptr<TransactionLogIterator> iter;
+  Status status = dbfull()->GetUpdatesSince(0, &iter);
+  ASSERT_TRUE(status.ok());
+  ASSERT_TRUE(iter->Valid());
+  ASSERT_TRUE(status.ok());
+  ASSERT_TRUE(iter->Valid());
+  int i = 0;
+  SequenceNumber lastSequence = 0;
+  while (iter->Valid()) {
+    BatchResult res = iter->GetBatch();
+    ASSERT_TRUE(res.sequence > lastSequence);
+    lastSequence = res.sequence;
+    ASSERT_TRUE(iter->status().ok());
+    iter->Next();
+    ++i;
+  }
+  ASSERT_EQ(i, 2);
+}
+
 TEST(DBTest, ReadCompaction) {
   std::string value(4096, '4'); // a string of size 4K
   {