Do not log unnecessary WAL obsoletion events (#7765)

Summary:
min_wal_number_to_keep should not be decreasing, if it does not increase, then there is no need to log the WAL obsoletions in MANIFEST since a previous one has been logged.

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

Test Plan: watch existing tests and stress tests to pass

Reviewed By: pdillinger

Differential Revision: D25462542

Pulled By: cheng-chang

fbshipit-source-id: 0085fcb6edf5cf2b0fc32f9932a7566f508768ff

db/memtable_list.cc

@@ -481,16 +481,20 @@ Status MemTableList::TryInstallMemtableFlushResults(
         // We piggyback the information of  earliest log file to keep in the
         // manifest entry for the last file flushed.
         edit_list.back()->SetMinLogNumberToKeep(min_wal_number_to_keep);
-      } else {
-        min_wal_number_to_keep =
-            PrecomputeMinLogNumberToKeepNon2PC(vset, *cfd, edit_list);
       }
 
       std::unique_ptr<VersionEdit> wal_deletion;
       if (vset->db_options()->track_and_verify_wals_in_manifest) {
-        wal_deletion.reset(new VersionEdit);
-        wal_deletion->DeleteWalsBefore(min_wal_number_to_keep);
-        edit_list.push_back(wal_deletion.get());
+        if (!vset->db_options()->allow_2pc) {
+          min_wal_number_to_keep =
+              PrecomputeMinLogNumberToKeepNon2PC(vset, *cfd, edit_list);
+        }
+        if (min_wal_number_to_keep >
+            vset->GetWalSet().GetMinWalNumberToKeep()) {
+          wal_deletion.reset(new VersionEdit);
+          wal_deletion->DeleteWalsBefore(min_wal_number_to_keep);
+          edit_list.push_back(wal_deletion.get());
+        }
       }
 
       const auto manifest_write_cb = [this, cfd, batch_count, log_buffer,
@@ -754,10 +758,12 @@ Status InstallMemtableAtomicFlushResults(
       min_wal_number_to_keep =
           PrecomputeMinLogNumberToKeepNon2PC(vset, cfds, edit_lists);
     }
-    wal_deletion.reset(new VersionEdit);
-    wal_deletion->DeleteWalsBefore(min_wal_number_to_keep);
-    edit_lists.back().push_back(wal_deletion.get());
-    ++num_entries;
+    if (min_wal_number_to_keep > vset->GetWalSet().GetMinWalNumberToKeep()) {
+      wal_deletion.reset(new VersionEdit);
+      wal_deletion->DeleteWalsBefore(min_wal_number_to_keep);
+      edit_lists.back().push_back(wal_deletion.get());
+      ++num_entries;
+    }
   }
 
   // Mark the version edits as an atomic group if the number of version edits