From 02c459805bd6809e7309fcf56e7673d444b47dd5 Mon Sep 17 00:00:00 2001
From: Abhishek Kona <abhishekk@fb.com>
Date: Mon, 18 Mar 2013 14:50:59 -0700
Subject: [PATCH] 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 | 25 +++++++++++++++++++++++++
 db/db_impl.h  |  2 ++
 db/db_test.cc | 32 ++++++++++++++++++++++++++++++++
 3 files changed, 59 insertions(+)

diff --git a/db/db_impl.cc b/db/db_impl.cc
index e4443c757..940164ea5 100644
--- a/db/db_impl.cc
+++ b/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) {
diff --git a/db/db_impl.h b/db/db_impl.h
index e5a2936cb..92ad211d5 100644
--- a/db/db_impl.h
+++ b/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);
diff --git a/db/db_test.cc b/db/db_test.cc
index fcdb4917c..dc069a892 100644
--- a/db/db_test.cc
+++ b/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
   {