[sys-4816] handle rocksdb upgrade to 7.10 in backwards compatible way (#277)

A new per-file entity called epoch number was introduced in v7.10, which is used to sort L0 files (previously sorted by largest_seqno). We need to make sure this number matches in leader follower. This is quite tricky during leaf deploy.

* leader is on new version while follower is on old version. This is fine actually. epoch number will just be skipped on follower
* follower is on new version while leader is on old version. This is the tricky one and this entire PR is to deal with this case.
The hack we use is to make follower to ignore epoch number from leader but simply calculating the epoch number on the fly. There are two cases:

* flush which generates L0 files. epoch number is allocated based on next_epoch_number of each CF. The L0 files are sorted based on largest seqno.
* compaction which merges files in lower levels to higher levels. epoch number = min epoch number of input files.
This is mostly fine, except when db is reopened. Rocksdb doesn't track next_epoch_number in manifest file. When db is
reopened, it calculates the next_epoch_number based on max epoch number of existing live files. So it's possible for the next_epoch_number to go backwards when db is reopened. Following two cases are possible:

* leader reopens db, causing next_epoch_number on leader to go backwards. So follower needs to rewind it.
* follower reopens db, causing next_epoch_number on follower to go backwards. So follower needs to advance it.
A new replication option: AR_RESET_IF_EPOCH_MISMATCH is added to help with this issue. rewind and advance are handled carefully to make sure it doesn't break existing file ordering.

The change in this PR is quite hacky, but fortunately we can remove most of them once rocksdb is fully upgraded

Author: seckcoder

cloud/replication_test.cc

@@ -224,6 +224,11 @@ class ReplicationTest : public testing::Test {
   DB* currentLeader() const {
     return leader_db_.get();
   }
+
+  DBImpl* leaderFull() const {
+    return static_cast_with_check<DBImpl>(currentLeader());
+  }
+
   DB* currentFollower() const {
     return follower_db_.get();
   }
@@ -265,12 +270,39 @@ class ReplicationTest : public testing::Test {
       return keys;
   }
 
+  // verify that the current log structured merge tree of two CFs to be the same
+  void verifyLSMTEqual(ColumnFamilyHandle* h1, ColumnFamilyHandle* h2) {
+    auto cf1 = static_cast_with_check<ColumnFamilyHandleImpl>(h1)->cfd(),
+         cf2 = static_cast_with_check<ColumnFamilyHandleImpl>(h2)->cfd();
+    ASSERT_EQ(cf1->NumberLevels(), cf2->NumberLevels())
+        << h1->GetName() << ", " << h2->GetName();
+
+    for (int level = 0; level < cf1->NumberLevels(); level++) {
+        auto files1 = cf1->current()->storage_info()->LevelFiles(level),
+             files2 = cf2->current()->storage_info()->LevelFiles(level);
+        ASSERT_EQ(files1.size(), files2.size())
+          << "mismatched number of files at level: " << level
+          << " between cf: " << cf1->GetName()
+          << " and cf: " << cf2->GetName();
+        for (size_t i = 0; i < files1.size(); i++) {
+          auto f1 = files1[i], f2 = files2[i];
+          ASSERT_EQ(f1->fd.file_size, f2->fd.file_size);
+          ASSERT_EQ(f1->fd.smallest_seqno, f2->fd.smallest_seqno);
+          ASSERT_EQ(f1->fd.largest_seqno, f2->fd.largest_seqno);
+          ASSERT_EQ(f1->epoch_number, f2->epoch_number);
+          ASSERT_EQ(f1->file_checksum, f2->file_checksum);
+          ASSERT_EQ(f1->unique_id, f2->unique_id);
+        }
+    }
+  }
+
   void verifyEqual() {
     ASSERT_EQ(leader_cfs_.size(), follower_cfs_.size());
     auto leader = leader_db_.get(), follower = follower_db_.get();
     for (auto& [name, cf1]: leader_cfs_) {
       auto cf2 = followerCF(name);
       verifyNextLogNumAndReplSeqConsistency(name);
+      verifyLSMTEqual(cf1.get(), cf2);
 
       auto itrLeader = std::unique_ptr<Iterator>(
           leader->NewIterator(ReadOptions(), cf1.get()));
@@ -290,6 +322,7 @@ class ReplicationTest : public testing::Test {
 
 protected:
   std::shared_ptr<Logger> info_log_;
+  bool replicate_epoch_number_{true};
   void resetFollowerSequence(int new_seq) {
     followerSequence_ = new_seq;
   }
@@ -420,6 +453,12 @@ size_t ReplicationTest::catchUpFollower(
   MutexLock lock(&log_records_mutex_);
   DB::ApplyReplicationLogRecordInfo info;
   size_t ret = 0;
+  unsigned flags = DB::AR_EVICT_OBSOLETE_FILES;
+  if (replicate_epoch_number_) {
+    flags |= DB::AR_REPLICATE_EPOCH_NUM;
+  } else {
+    flags |= DB::AR_RESET_IF_EPOCH_MISMATCH;
+  }
   for (; followerSequence_ < (int)log_records_.size(); ++followerSequence_) {
     if (num_records && ret >= *num_records) {
       break;
@@ -430,8 +469,9 @@ size_t ReplicationTest::catchUpFollower(
         [this](Slice) {
           return ColumnFamilyOptions(follower_db_->GetOptions());
         },
-        allow_new_manifest_writes, &info, DB::AR_EVICT_OBSOLETE_FILES);
+        allow_new_manifest_writes, &info, flags);
     assert(s.ok());
+    assert(info.mismatched_epoch_num == 0);
     ++ret;
   }
   if (info.has_new_manifest_writes) {
@@ -1098,7 +1138,18 @@ TEST_F(ReplicationTest, EvictObsoleteFiles) {
       static_cast_with_check<DBImpl>(follower)->TEST_table_cache()->GetUsage());
 }
 
-TEST_F(ReplicationTest, Stress) {
+class ReplicationTestWithParam : public ReplicationTest,
+                                 public testing::WithParamInterface<bool> {
+ public:
+  ReplicationTestWithParam()
+    : ReplicationTest() {}
+
+  void SetUp() override {
+    replicate_epoch_number_ = GetParam();
+  }
+};
+
+TEST_P(ReplicationTestWithParam, Stress) {
   std::string val;
   auto leader = openLeader();
   openFollower();
@@ -1114,49 +1165,74 @@ TEST_F(ReplicationTest, Stress) {
     createColumnFamily(cf(i));
   }
 
-  auto do_writes = [&](int n) {
-    auto rand = Random::GetTLSInstance();
-    while (n > 0) {
-      auto cfi = rand->Uniform(kColumnFamilyCount);
-      rocksdb::WriteBatch wb;
-      for (size_t i = 0; i < 3; ++i) {
-        --n;
-        wb.Put(leaderCF(cf(cfi)), std::to_string(rand->Uniform(kMaxKey)),
-               std::to_string(rand->Next()));
+  auto do_writes = [&]() {
+    auto writes_per_thread = [&](int n) {
+      auto rand = Random::GetTLSInstance();
+      while (n > 0) {
+        auto cfi = rand->Uniform(kColumnFamilyCount);
+        rocksdb::WriteBatch wb;
+        for (size_t i = 0; i < 3; ++i) {
+          --n;
+          wb.Put(leaderCF(cf(cfi)), std::to_string(rand->Uniform(kMaxKey)),
+                 std::to_string(rand->Next()));
+        }
+        ASSERT_OK(leader->Write(wo(), &wb));
       }
-      ASSERT_OK(leader->Write(wo(), &wb));
+    };
+
+    std::vector<std::thread> threads;
+    for (size_t i = 0; i < kThreadCount; ++i) {
+      threads.emplace_back([&]() { writes_per_thread(kWritesPerThread); });
+    }
+    for (auto& t : threads) {
+      t.join();
     }
+
+    ASSERT_OK(
+      leaderFull()->TEST_WaitForBackgroundWork());
   };
 
-  std::vector<std::thread> threads;
-  for (size_t i = 0; i < kThreadCount; ++i) {
-    threads.emplace_back([&]() { do_writes(kWritesPerThread); });
-  }
-  for (auto& t : threads) {
-    t.join();
-  }
-  ASSERT_OK(
-      static_cast_with_check<DBImpl>(leader)->TEST_WaitForBackgroundWork());
+  auto verifyNextEpochNumber = [&]() {
+    for (int i = 0; i < kColumnFamilyCount; i++) {
+      auto cf1 = leaderCFD(cf(i)), cf2 = followerCFD(cf(i));
+      ASSERT_EQ(cf1->GetNextEpochNumber(), cf2->GetNextEpochNumber());
+    }
+  };
 
-  catchUpFollower();
+  do_writes();
 
+  catchUpFollower();
   verifyEqual();
+  verifyNextEpochNumber();
+
+  ROCKS_LOG_INFO(info_log_, "reopen leader");
 
   // Reopen leader
   closeLeader();
   leader = openLeader();
-  ASSERT_OK(leader->Flush(FlushOptions()));
-
+  // memtable might not be empty after reopening leader, since we recover
+  // replication log when opening it.
+  ASSERT_OK(leader->Flush({}));
+  ASSERT_OK(leaderFull()->TEST_WaitForBackgroundWork());
+  catchUpFollower();
   verifyEqual();
 
+  do_writes();
+
+  ROCKS_LOG_INFO(info_log_, "reopen follower");
+
   // Reopen follower
   closeFollower();
   openFollower();
   catchUpFollower();
 
   verifyEqual();
+  verifyNextEpochNumber();
 }
 
+INSTANTIATE_TEST_CASE_P(ReplicationTest, ReplicationTestWithParam,
+                        ::testing::Values(false, true));
+
 TEST_F(ReplicationTest, DeleteRange) {
   auto leader = openLeader();
   openFollower();
@@ -1201,6 +1277,26 @@ TEST_F(ReplicationTest, DeleteRange) {
   verifyEqual();
 }
 
+TEST_F(ReplicationTest, EpochNumberSimple) {
+  auto options = leaderOptions();
+  options.disable_auto_compactions = true;
+  auto leader = openLeader();
+  openFollower();
+
+  ASSERT_OK(leader->Put(wo(), "k1", "v1"));
+  ASSERT_OK(leader->Flush({}));
+  catchUpFollower();
+
+  ASSERT_OK(leader->Put(wo(), "k1", "v2"));
+  ASSERT_OK(leader->Flush({}));
+  auto leaderFull = static_cast_with_check<DBImpl>(leader);
+  ASSERT_OK(leaderFull->TEST_CompactRange(0, nullptr, nullptr, nullptr, true));
+
+  catchUpFollower();
+
+  verifyEqual();
+}
+
 }  //  namespace ROCKSDB_NAMESPACE
 
 // A black-box test for the cloud wrapper around rocksdb

db/compaction/compaction_job.cc

@@ -1590,11 +1590,12 @@ Status CompactionJob::FinishCompactionOutputFile(
     outputs.UpdateTableProperties();
     ROCKS_LOG_INFO(db_options_.info_log,
                    "[%s] [JOB %d] Generated table #%" PRIu64 ": %" PRIu64
-                   " keys, %" PRIu64 " bytes%s, temperature: %s",
+                   " keys, %" PRIu64 " bytes%s, temperature: %s, epoch number: %" PRIu64,
                    cfd->GetName().c_str(), job_id_, output_number,
                    current_entries, meta->fd.file_size,
                    meta->marked_for_compaction ? " (need compaction)" : "",
-                   temperature_to_string[meta->temperature].c_str());
+                   temperature_to_string[meta->temperature].c_str(),
+                   meta->epoch_number);
   }
   std::string fname;
   FileDescriptor output_fd;

db/db_impl/db_impl.cc

@@ -1217,6 +1217,7 @@ std::string DescribeVersionEdit(const VersionEdit& e, ColumnFamilyData* cfd) {
       }
       first = false;
       oss << f.second.fd.GetNumber();
+      oss << ":" << f.second.epoch_number;
     }
     oss << "] ";
   }
@@ -1424,9 +1425,132 @@ Status DBImpl::ApplyReplicationLogRecord(ReplicationLogRecord record,
           autovector<VersionEdit*> el;
           el.push_back(&e);
           edit_lists.push_back(std::move(el));
-
           ROCKS_LOG_INFO(immutable_db_options_.info_log, "%s",
                          DescribeVersionEdit(e, cfd).c_str());
+          auto& newFiles = e.GetNewFiles();
+          bool epoch_recovery_succeeded = true;
+          std::ostringstream err_oss;
+          if (!(flags & AR_REPLICATE_EPOCH_NUM)) {
+            // Epoch number calculation on the fly.
+            // There are two cases in which we need to calculate epoch number
+            // when applying `kManifestWrite`
+            // 1. flush which generates L0 files. epoch number is allocated
+            // based on `next_epoch_number` of each CF. The L0 files are sorted
+            // based on `largest seqno`. 
+            // 2. compaction which merges files in lower levels to higher
+            // levels. epoch number = min epoch number of input files.
+            const auto& deletedFiles = e.GetDeletedFiles();
+            if (deletedFiles.empty() && !newFiles.empty()) {
+              // case 1: flush into L0 files. New files must be level 0
+
+              for (auto& p : newFiles) {
+                if (p.first != 0) {
+                  epoch_recovery_succeeded = false;
+                  err_oss << "newly flushed file: " << p.first << " is not at L0";
+                  break;
+                }
+              }
+
+              // sort added files by largest seqno
+              std::vector<FileMetaData*> added_files;
+              for(auto& p: newFiles) {
+                added_files.push_back(&p.second);
+              }
+
+              NewestFirstBySeqNo cmp;
+              std::sort(added_files.begin(), added_files.end(), cmp);
+              auto first_file = added_files[0];
+              // Rewind/advance next_epoch_number. This is necessary if epoch_number
+              // mismtaches due to db reopen.
+              if (first_file->epoch_number != kUnknownEpochNumber &&
+                  first_file->epoch_number != cfd->GetNextEpochNumber() &&
+                  (flags & AR_RESET_IF_EPOCH_MISMATCH)) {
+                auto max_epoch_number =
+                    cfd->current()->storage_info()->GetMaxEpochNumberOfFiles();
+                if (first_file->epoch_number < cfd->GetNextEpochNumber() &&
+                    (first_file->epoch_number == max_epoch_number + 1)) {
+                  ROCKS_LOG_INFO(immutable_db_options_.info_log,
+                                 "[%s] rewind next_epoch_number from: %" PRIu64
+                                 " to %" PRIu64,
+                                 cfd->GetName().c_str(),
+                                 cfd->GetNextEpochNumber(),
+                                 max_epoch_number + 1);
+                  cfd->SetNextEpochNumber(max_epoch_number + 1);
+                } else if (first_file->epoch_number >
+                               cfd->GetNextEpochNumber() &&
+                           (cfd->GetNextEpochNumber() ==
+                            max_epoch_number + 1)) {
+                  ROCKS_LOG_INFO(immutable_db_options_.info_log,
+                                 "[%s] advance next_epoch_number from: %" PRIu64
+                                 " to %" PRIu64,
+                                 cfd->GetName().c_str(),
+                                 cfd->GetNextEpochNumber(),
+                                 first_file->epoch_number);
+                  cfd->SetNextEpochNumber(first_file->epoch_number);
+                } else {
+                  ROCKS_LOG_ERROR(immutable_db_options_.info_log,
+                                  "[%s] unexpected epoch number: %" PRIu64
+                                  " for file: %" PRIu64
+                                  " ; max epoch number: %" PRIu64,
+                                  cfd->GetName().c_str(),
+                                  first_file->epoch_number,
+                                  first_file->fd.GetNumber(),
+                                  max_epoch_number);
+                  s = Status::Corruption("unexpected epoch number for added file");
+                  break;
+                }
+              }
+
+              for (auto meta: added_files) {
+                auto old_epoch_number = meta->epoch_number;
+                meta->epoch_number = cfd->NewEpochNumber();
+                if (old_epoch_number != meta->epoch_number) {
+                  info->mismatched_epoch_num += 1;
+                }
+              }
+            } else if (!deletedFiles.empty() && !newFiles.empty()) {
+              // case 2: compaction
+              uint64_t min_input_epoch_number =
+                  std::numeric_limits<uint64_t>::max();
+              const auto& storage_info = cfd->current()->storage_info();
+              for (auto [level, file_number] : deletedFiles) {
+                auto meta = storage_info->GetFileMetaDataByNumber(file_number);
+                if (!meta) {
+                  err_oss << "deleted file: " << file_number
+                          << " at level: " << level << " not found";
+                  break;
+                }
+                min_input_epoch_number =
+                    std::min(meta->epoch_number, min_input_epoch_number);
+              }
+
+              for (auto& p: newFiles) {
+                auto old_epoch_number = p.second.epoch_number;
+                p.second.epoch_number = min_input_epoch_number;
+                if (old_epoch_number != p.second.epoch_number) {
+                  info->mismatched_epoch_num += 1;
+                }
+              }
+            }
+          } else if (newFiles.size() > 0) {
+            // Maintain next epoch number on follower
+            auto next_epoch_number = cfd->GetNextEpochNumber();
+            for (auto& p : newFiles) {
+              auto epoch_number = p.second.epoch_number;
+              // advance next epoch number. next_epoch_number never goes
+              // backwards
+              if (epoch_number != kUnknownEpochNumber &&
+                  (epoch_number >= next_epoch_number)) {
+                next_epoch_number = epoch_number + 1;
+              }
+            }
+            cfd->SetNextEpochNumber(next_epoch_number);
+          }
+
+          if (!epoch_recovery_succeeded) {
+            s = Status::Corruption(err_oss.str());
+            break;
+          }
         }
         if (!s.ok()) {
           break;

db/flush_job.cc

@@ -895,9 +895,9 @@ Status FlushJob::WriteLevel0Table() {
           NewMergingIterator(&cfd_->internal_comparator(), memtables.data(),
                              static_cast<int>(memtables.size()), &arena));
       ROCKS_LOG_INFO(db_options_.info_log,
-                     "[%s] [JOB %d] Level-0 flush table #%" PRIu64 ": started",
+                     "[%s] [JOB %d] Level-0 flush table #%" PRIu64 ": started. Epoch number: %" PRIu64,
                      cfd_->GetName().c_str(), job_context_->job_id,
-                     meta_.fd.GetNumber());
+                     meta_.fd.GetNumber(), meta_.epoch_number);
 
       TEST_SYNC_POINT_CALLBACK("FlushJob::WriteLevel0Table:output_compression",
                                &output_compression_);