From 11c4be2222e9a26c43a965b907b30d4c8e269a67 Mon Sep 17 00:00:00 2001
From: Yanqin Jin <yanqin@fb.com>
Date: Tue, 8 Dec 2020 02:36:18 -0800
Subject: [PATCH] Refactor ProcessManifestWrites a little bit (#7751)

Summary:
This PR removes a nested loop inside ProcessManifestWrites. The new
implementation has the same behavior as the old code with simpler logic
and lower complexity.

Pull Request resolved: https://github.com/facebook/rocksdb/pull/7751

Test Plan:
make check
Run make crash_test on devserver and succeeds 3 times.

Reviewed By: ltamasi

Differential Revision: D25363526

Pulled By: riversand963

fbshipit-source-id: 27e681949dacd7501a752e5e517b9e85b54ccb2e
---
 db/version_set.cc | 38 +++++++++++++++-----------------------
 1 file changed, 15 insertions(+), 23 deletions(-)

diff --git a/db/version_set.cc b/db/version_set.cc
index 2f5348cfb..c77a85e59 100644
--- a/db/version_set.cc
+++ b/db/version_set.cc
@@ -4229,31 +4229,23 @@ Status VersionSet::ProcessManifestWrites(
       // Each version in versions corresponds to a column family.
       // For each column family, update its log number indicating that logs
       // with number smaller than this should be ignored.
-      // TODO (yanqin): remove the nested loop if possible.
-      for (const auto version : versions) {
-        uint64_t max_log_number_in_batch = 0;
-        assert(version->cfd_);
-        uint32_t cf_id = version->cfd_->GetID();
-        std::string full_history_ts_low;
-        for (const auto& e : batch_edits) {
-          if (e->column_family_ == cf_id) {
-            if (e->has_log_number_) {
-              max_log_number_in_batch =
-                  std::max(max_log_number_in_batch, e->log_number_);
-            }
-            if (e->HasFullHistoryTsLow()) {
-              version->cfd_->SetFullHistoryTsLow(e->GetFullHistoryTsLow());
-            }
-          }
+      uint64_t last_min_log_number_to_keep = 0;
+      for (const auto& e : batch_edits) {
+        ColumnFamilyData* cfd = nullptr;
+        if (!e->IsColumnFamilyManipulation()) {
+          cfd = column_family_set_->GetColumnFamily(e->column_family_);
+          // e would not have been added to batch_edits if its corresponding
+          // column family is dropped.
+          assert(cfd);
         }
-        if (max_log_number_in_batch != 0) {
-          assert(version->cfd_->GetLogNumber() <= max_log_number_in_batch);
-          version->cfd_->SetLogNumber(max_log_number_in_batch);
+        if (cfd) {
+          if (e->has_log_number_ && e->log_number_ > cfd->GetLogNumber()) {
+            cfd->SetLogNumber(e->log_number_);
+          }
+          if (e->HasFullHistoryTsLow()) {
+            cfd->SetFullHistoryTsLow(e->GetFullHistoryTsLow());
+          }
         }
-      }
-
-      uint64_t last_min_log_number_to_keep = 0;
-      for (auto& e : batch_edits) {
         if (e->has_min_log_number_to_keep_) {
           last_min_log_number_to_keep =
               std::max(last_min_log_number_to_keep, e->min_log_number_to_keep_);