From d854abad78a61d939e6dd267d9b4a811946ae41d Mon Sep 17 00:00:00 2001
From: Levi Tamasi <ltamasi@fb.com>
Date: Thu, 11 Jun 2020 18:29:51 -0700
Subject: [PATCH] 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 | 44 +++++++++++++++++++++++++++++--------------
 1 file changed, 30 insertions(+), 14 deletions(-)

diff --git a/db/version_builder.cc b/db/version_builder.cc
index d614ee8f9..9c44d9d16 100644
--- a/db/version_builder.cc
+++ b/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_);
-      vstorage->AddFile(level, f, ioptions_->info_log);
+      const auto& add_files = level_state.added_files;
+      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);
+      }
     }
   }
 };