Fix recovery for WALs without data for all CFs

Summary:
if one or more CFs had no data in the WAL, the log number that's used
by FindObsoleteFiles() wasn't updated. We need to treat this case the same as
if the data for that WAL had been flushed.

Test Plan: new unit test

Reviewers: IslamAbdelRahman, yiwu, sdong

Reviewed By: sdong

Subscribers: andrewkr, dhruba, leveldb

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

diff --git a/db/db_impl.cc b/db/db_impl.cc
index 126c3e943e05613e04a472dc18c104c289e2dc5a..f750ef8390b2e2df5071c0f8c1b4c0948991cd47 100644 (file)
--- a/db/db_impl.cc
+++ b/db/db_impl.cc
@@ -1701,7 +1701,7 @@ Status DBImpl::RecoverLogFiles(const std::vector<uint64_t>& log_numbers,
       // recovered and should be ignored on next reincarnation.
       // Since we already recovered max_log_number, we want all logs
       // with numbers `<= max_log_number` (includes this one) to be ignored
-      if (flushed) {
+      if (flushed || cfd->mem()->GetFirstSequenceNumber() == 0) {
         edit->SetLogNumber(max_log_number + 1);
       }
       // we must mark the next log number as used, even though it's

diff --git a/db/db_wal_test.cc b/db/db_wal_test.cc
index 34d780511078b4185423235c48de27b635d096d0..10520b774a7489831a2f5cdf5936bbacaf5d599b 100644 (file)
--- a/db/db_wal_test.cc
+++ b/db/db_wal_test.cc
@@ -305,6 +305,31 @@ TEST_F(DBWALTest, GetSortedWalFiles) {
   } while (ChangeOptions());
 }
 
+TEST_F(DBWALTest, RecoveryWithLogDataForSomeCFs) {
+  // Test for regression of WAL cleanup missing files that don't contain data
+  // for every column family.
+  do {
+    CreateAndReopenWithCF({"pikachu"}, CurrentOptions());
+    ASSERT_OK(Put(1, "foo", "v1"));
+    ASSERT_OK(Put(1, "foo", "v2"));
+    std::array<uint64_t, 2> earliest_log_nums;
+    for (int i = 0; i < 2; ++i) {
+      if (i > 0) {
+        ReopenWithColumnFamilies({"default", "pikachu"}, CurrentOptions());
+      }
+      VectorLogPtr log_files;
+      ASSERT_OK(dbfull()->GetSortedWalFiles(log_files));
+      if (log_files.size() > 0) {
+        earliest_log_nums[i] = log_files[0]->LogNumber();
+      } else {
+        earliest_log_nums[i] = port::kMaxUint64;
+      }
+    }
+    // Check at least the first WAL was cleaned up during the recovery.
+    ASSERT_LT(earliest_log_nums[0], earliest_log_nums[1]);
+  } while (ChangeOptions());
+}
+
 TEST_F(DBWALTest, RecoverWithLargeLog) {
   do {
     {