From 55ab150b46018550ccd86ea31c3968a627ed94ca Mon Sep 17 00:00:00 2001
From: Andrew Kryczka <andrewkr@fb.com>
Date: Thu, 15 Sep 2016 09:55:02 -0700
Subject: [PATCH] Fix GetSortedWalFiles when log recycling enabled

Summary:
Previously the sequence number was mistakenly passed in an argument
where the log number should go. This caused the reader to assume the old WAL
format was used, which is incompatible with the WAL recycling format.

Test Plan:
new unit test, verified it fails before this change and passes
afterwards.

Reviewers: yiwu, lightmark, IslamAbdelRahman, sdong

Reviewed By: sdong

Subscribers: andrewkr, dhruba, leveldb

Differential Revision: https://reviews.facebook.net/D63987
---
 db/db_wal_test.cc      | 13 +++++++++++++
 db/wal_manager.cc      |  7 ++++---
 db/wal_manager.h       |  7 ++++---
 db/wal_manager_test.cc |  5 +++--
 4 files changed, 24 insertions(+), 8 deletions(-)

diff --git a/db/db_wal_test.cc b/db/db_wal_test.cc
index 868d240d9..34d780511 100644
--- a/db/db_wal_test.cc
+++ b/db/db_wal_test.cc
@@ -292,6 +292,19 @@ TEST_F(DBWALTest, RecoveryWithEmptyLog) {
 }
 
 #ifndef ROCKSDB_LITE
+TEST_F(DBWALTest, GetSortedWalFiles) {
+  do {
+    CreateAndReopenWithCF({"pikachu"}, CurrentOptions());
+    VectorLogPtr log_files;
+    ASSERT_OK(dbfull()->GetSortedWalFiles(log_files));
+    ASSERT_EQ(0, log_files.size());
+
+    ASSERT_OK(Put(1, "foo", "v1"));
+    ASSERT_OK(dbfull()->GetSortedWalFiles(log_files));
+    ASSERT_EQ(1, log_files.size());
+  } while (ChangeOptions());
+}
+
 TEST_F(DBWALTest, RecoverWithLargeLog) {
   do {
     {
diff --git a/db/wal_manager.cc b/db/wal_manager.cc
index da4f33aab..e57c12040 100644
--- a/db/wal_manager.cc
+++ b/db/wal_manager.cc
@@ -383,7 +383,7 @@ Status WalManager::ReadFirstRecord(const WalFileType type,
   Status s;
   if (type == kAliveLogFile) {
     std::string fname = LogFileName(db_options_.wal_dir, number);
-    s = ReadFirstLine(fname, sequence);
+    s = ReadFirstLine(fname, number, sequence);
     if (env_->FileExists(fname).ok() && !s.ok()) {
       // return any error that is not caused by non-existing file
       return s;
@@ -394,7 +394,7 @@ Status WalManager::ReadFirstRecord(const WalFileType type,
     //  check if the file got moved to archive.
     std::string archived_file =
         ArchivedLogFileName(db_options_.wal_dir, number);
-    s = ReadFirstLine(archived_file, sequence);
+    s = ReadFirstLine(archived_file, number, sequence);
     // maybe the file was deleted from archive dir. If that's the case, return
     // Status::OK(). The caller with identify this as empty file because
     // *sequence == 0
@@ -413,6 +413,7 @@ Status WalManager::ReadFirstRecord(const WalFileType type,
 // the function returns status.ok() and sequence == 0 if the file exists, but is
 // empty
 Status WalManager::ReadFirstLine(const std::string& fname,
+                                 const uint64_t number,
                                  SequenceNumber* sequence) {
   struct LogReporter : public log::Reader::Reporter {
     Env* env;
@@ -449,7 +450,7 @@ Status WalManager::ReadFirstLine(const std::string& fname,
   reporter.status = &status;
   reporter.ignore_error = !db_options_.paranoid_checks;
   log::Reader reader(db_options_.info_log, std::move(file_reader), &reporter,
-                     true /*checksum*/, 0 /*initial_offset*/, *sequence);
+                     true /*checksum*/, 0 /*initial_offset*/, number);
   std::string scratch;
   Slice record;
 
diff --git a/db/wal_manager.h b/db/wal_manager.h
index a3079ed48..d27985b9d 100644
--- a/db/wal_manager.h
+++ b/db/wal_manager.h
@@ -54,9 +54,9 @@ class WalManager {
     return ReadFirstRecord(type, number, sequence);
   }
 
-  Status TEST_ReadFirstLine(const std::string& fname,
+  Status TEST_ReadFirstLine(const std::string& fname, const uint64_t number,
                             SequenceNumber* sequence) {
-    return ReadFirstLine(fname, sequence);
+    return ReadFirstLine(fname, number, sequence);
   }
 
  private:
@@ -71,7 +71,8 @@ class WalManager {
   Status ReadFirstRecord(const WalFileType type, const uint64_t number,
                          SequenceNumber* sequence);
 
-  Status ReadFirstLine(const std::string& fname, SequenceNumber* sequence);
+  Status ReadFirstLine(const std::string& fname, const uint64_t number,
+                       SequenceNumber* sequence);
 
   // ------- state from DBImpl ------
   const DBOptions& db_options_;
diff --git a/db/wal_manager_test.cc b/db/wal_manager_test.cc
index 5db538a31..9c7ac50ba 100644
--- a/db/wal_manager_test.cc
+++ b/db/wal_manager_test.cc
@@ -119,10 +119,11 @@ TEST_F(WalManagerTest, ReadFirstRecordCache) {
   ASSERT_OK(env_->NewWritableFile(path, &file, EnvOptions()));
 
   SequenceNumber s;
-  ASSERT_OK(wal_manager_->TEST_ReadFirstLine(path, &s));
+  ASSERT_OK(wal_manager_->TEST_ReadFirstLine(path, 1 /* number */, &s));
   ASSERT_EQ(s, 0U);
 
-  ASSERT_OK(wal_manager_->TEST_ReadFirstRecord(kAliveLogFile, 1, &s));
+  ASSERT_OK(
+      wal_manager_->TEST_ReadFirstRecord(kAliveLogFile, 1 /* number */, &s));
   ASSERT_EQ(s, 0U);
 
   unique_ptr<WritableFileWriter> file_writer(