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

@@ -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 {
     {

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;
 

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_;

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(