improve consistency checks for deleted files (#189) (#191)

This PR optimizes consistency checks performance for the case with a lot of deleted files. 

When doing `DeleteFilesInRange` with `force_consistency_checks` on for a large range which spans about 16.5K SST files, we found it spent most of the time in `LogAndApply`(the time in red rectangle).
![image](https://user-images.githubusercontent.com/13497871/89766469-c8ab4b80-db2a-11ea-9064-9c0a969e5453.png)

In `CheckConsistencyForDeletes`, it traverses the whole LSM to check if the file existed in the previous version for every deleted file. In the case of a lot of deleted files such as `DeleteFilesInRange`, it would waste too much time on this operation. Even worse, it is done with db mutex held, so it may greatly affect the foreground write performance.

After making the check in batch with only one round of traverse of the LSM, the time is greatly reduced.
![image](https://user-images.githubusercontent.com/13497871/89766495-d1038680-db2a-11ea-961b-cb009e93cdd6.png)

Signed-off-by: Connor1996 <zbk602423539@gmail.com>

db/version_builder.cc

@@ -220,52 +220,65 @@ class VersionBuilder::Rep {
     return Status::OK();
   }
 
-  Status CheckConsistencyForDeletes(VersionEdit* /*edit*/, uint64_t number,
-                                    int level) {
+  Status CheckConsistencyForDeletes(VersionEdit* edit) {
 #ifdef NDEBUG
     if (!base_vstorage_->force_consistency_checks()) {
       // Dont run consistency checks in release mode except if
       // explicitly asked to
       return Status::OK();
     }
 #endif
+    std::unordered_map<uint64_t, int> deletes;
+    for (const auto& del_file : edit->GetDeletedFiles()) {
+      const auto level = del_file.first;
+      const auto number = del_file.second;
+      if (level < num_levels_) {
+        deletes[number] = level;
+      }
+    }
+
     // a file to be deleted better exist in the previous version
-    bool found = false;
-    for (int l = 0; !found && l < num_levels_; l++) {
+    for (int l = 0; deletes.size() != 0 && l < num_levels_; l++) {
       const std::vector<FileMetaData*>& base_files =
           base_vstorage_->LevelFiles(l);
       for (size_t i = 0; i < base_files.size(); i++) {
         FileMetaData* f = base_files[i];
-        if (f->fd.GetNumber() == number) {
+        if (deletes.erase(f->fd.GetNumber()) != 0) {
+          if (deletes.size() == 0) {
+            break;
+          }
+        }
+      }
+    }
+
+    for (const auto& d : deletes) {
+      const auto number = d.first;
+      const auto level = d.second;
+      // if the file did not exist in the previous version, then it
+      // is possibly moved from lower level to higher level in current
+      // version
+      bool found = false;
+      for (int l = level + 1; l < num_levels_; l++) {
+        auto& level_added = levels_[l].added_files;
+        auto got = level_added.find(number);
+        if (got != level_added.end()) {
           found = true;
           break;
         }
       }
-    }
-    // if the file did not exist in the previous version, then it
-    // is possibly moved from lower level to higher level in current
-    // version
-    for (int l = level + 1; !found && l < num_levels_; l++) {
-      auto& level_added = levels_[l].added_files;
-      auto got = level_added.find(number);
-      if (got != level_added.end()) {
-        found = true;
-        break;
+      // maybe this file was added in a previous edit that was Applied
+      if (!found) {
+        auto& level_added = levels_[level].added_files;
+        auto got = level_added.find(number);
+        if (got != level_added.end()) {
+          found = true;
+        }
       }
-    }
-
-    // maybe this file was added in a previous edit that was Applied
-    if (!found) {
-      auto& level_added = levels_[level].added_files;
-      auto got = level_added.find(number);
-      if (got != level_added.end()) {
-        found = true;
+      if (!found) {
+        fprintf(stderr, "not found %" PRIu64 "\n", number);
+        return Status::Corruption("not found " + NumberToString(number));
       }
     }
-    if (!found) {
-      fprintf(stderr, "not found %" PRIu64 "\n", number);
-      return Status::Corruption("not found " + NumberToString(number));
-    }
     return Status::OK();
   }
 
@@ -291,22 +304,24 @@ class VersionBuilder::Rep {
 
     // Delete files
     const VersionEdit::DeletedFileSet& del = edit->GetDeletedFiles();
+    s = CheckConsistencyForDeletes(edit);
+    if (!s.ok()) {
+      return s;
+    }
     for (const auto& del_file : del) {
       const auto level = del_file.first;
       const auto number = del_file.second;
       if (level < num_levels_) {
         levels_[level].deleted_files.insert(number);
-        CheckConsistencyForDeletes(edit, number, level);
-
-        auto exising = levels_[level].added_files.find(number);
-        if (exising != levels_[level].added_files.end()) {
-          UnrefFile(exising->second);
-          levels_[level].added_files.erase(exising);
+        auto existing = levels_[level].added_files.find(number);
+        if (existing != levels_[level].added_files.end()) {
+          UnrefFile(existing->second);
+          levels_[level].added_files.erase(existing);
         }
       } else {
-        auto exising = invalid_levels_[level].find(number);
-        if (exising != invalid_levels_[level].end()) {
-          invalid_levels_[level].erase(exising);
+        auto existing = invalid_levels_[level].find(number);
+        if (existing != invalid_levels_[level].end()) {
+          invalid_levels_[level].erase(existing);
         } else {
           // Deleting an non-existing file on invalid level.
           has_invalid_levels_ = true;
@@ -515,9 +530,8 @@ Status VersionBuilder::CheckConsistency(VersionStorageInfo* vstorage) {
   return rep_->CheckConsistency(vstorage);
 }
 
-Status VersionBuilder::CheckConsistencyForDeletes(VersionEdit* edit,
-                                                  uint64_t number, int level) {
-  return rep_->CheckConsistencyForDeletes(edit, number, level);
+Status VersionBuilder::CheckConsistencyForDeletes(VersionEdit* edit) {
+  return rep_->CheckConsistencyForDeletes(edit);
 }
 
 bool VersionBuilder::CheckConsistencyForNumLevels() {

db/version_builder.h

@@ -8,6 +8,8 @@
 // found in the LICENSE file. See the AUTHORS file for names of contributors.
 //
 #pragma once
+#include <unordered_map>
+
 #include "rocksdb/env.h"
 #include "rocksdb/slice_transform.h"
 
@@ -28,8 +30,7 @@ class VersionBuilder {
                  VersionStorageInfo* base_vstorage, Logger* info_log = nullptr);
   ~VersionBuilder();
   Status CheckConsistency(VersionStorageInfo* vstorage);
-  Status CheckConsistencyForDeletes(VersionEdit* edit, uint64_t number,
-                                    int level);
+  Status CheckConsistencyForDeletes(VersionEdit* edit);
   bool CheckConsistencyForNumLevels();
   Status Apply(VersionEdit* edit);
   Status SaveTo(VersionStorageInfo* vstorage);