Revisit the handling of the case when a file is re-added to the same level (#6939)

Summary:
https://github.com/facebook/rocksdb/pull/6901 subtly changed the handling of the corner case
when a table file is deleted from a level, then re-added to the same level. (Note: this
should be extremely rare; one scenario that comes to mind is a trivial move followed by
a call to `ReFitLevel` that moves the file back to the original level.) Before that change,
a new `FileMetaData` object was created as a result of this sequence; after the change,
the original `FileMetaData` was essentially resurrected (since the deletion and the addition
simply cancel each other out with the change). This patch restores the original behavior,
which is more intuitive considering the interface, and in sync with how trivial moves are handled.
(Also note that `FileMetaData` contains some mutable data members, the values of which
might be different in the resurrected object and the freshly created one.)
The PR also fixes a bug in this area: with the original pre-6901 code, `VersionBuilder`
would add the same file twice to the same level in the scenario described above.
Pull Request resolved: https://github.com/facebook/rocksdb/pull/6939

Test Plan: `make check`

Reviewed By: ajkr

Differential Revision: D21905580

Pulled By: ltamasi

fbshipit-source-id: da07ae45384ecf3c6c53506d106432d88a7ec9df

db/version_builder.cc

@@ -470,12 +470,12 @@ class VersionBuilder::Rep {
     if (add_it != add_files.end()) {
       UnrefFile(add_it->second);
       add_files.erase(add_it);
-    } else {
-      auto& del_files = level_state.deleted_files;
-      assert(del_files.find(file_number) == del_files.end());
-      del_files.emplace(file_number);
     }
 
+    auto& del_files = level_state.deleted_files;
+    assert(del_files.find(file_number) == del_files.end());
+    del_files.emplace(file_number);
+
     table_file_levels_[file_number] =
         VersionStorageInfo::FileLocation::Invalid().GetLevel();
 
@@ -514,15 +514,15 @@ class VersionBuilder::Rep {
     auto del_it = del_files.find(file_number);
     if (del_it != del_files.end()) {
       del_files.erase(del_it);
-    } else {
-      FileMetaData* const f = new FileMetaData(meta);
-      f->refs = 1;
-
-      auto& add_files = level_state.added_files;
-      assert(add_files.find(file_number) == add_files.end());
-      add_files.emplace(file_number, f);
     }
 
+    FileMetaData* const f = new FileMetaData(meta);
+    f->refs = 1;
+
+    auto& add_files = level_state.added_files;
+    assert(add_files.find(file_number) == add_files.end());
+    add_files.emplace(file_number, f);
+
     table_file_levels_[file_number] = level;
 
     return Status::OK();
@@ -850,12 +850,28 @@ class VersionBuilder::Rep {
   }
 
   void MaybeAddFile(VersionStorageInfo* vstorage, int level, FileMetaData* f) {
-    if (levels_[level].deleted_files.count(f->fd.GetNumber()) > 0) {
+    const uint64_t file_number = f->fd.GetNumber();
+
+    const auto& level_state = levels_[level];
+
+    const auto& del_files = level_state.deleted_files;
+    const auto del_it = del_files.find(file_number);
+
+    if (del_it != del_files.end()) {
       // f is to-be-deleted table file
       vstorage->RemoveCurrentStats(f);
     } else {
-      assert(ioptions_);
+      const auto& add_files = level_state.added_files;
-      vstorage->AddFile(level, f, ioptions_->info_log);
+      const auto add_it = add_files.find(file_number);
+
+      // Note: if the file appears both in the base version and in the added
+      // list, the added FileMetaData supersedes the one in the base version.
+      if (add_it != add_files.end() && add_it->second != f) {
+        vstorage->RemoveCurrentStats(f);
+      } else {
+        assert(ioptions_);
+        vstorage->AddFile(level, f, ioptions_->info_log);
+      }
     }
   }
 };