From 6743135ea1bac252007ce9f83fb38f80f975ede1 Mon Sep 17 00:00:00 2001
From: Islam AbdelRahman <tec@fb.com>
Date: Tue, 1 Mar 2016 12:05:29 -0800
Subject: [PATCH] Fix DB::AddFile() issue when PurgeObsoleteFiles() is called

Summary:
In some situations the DB will scan all existing files in the DB path and delete the ones that are Obsolete.
If this happen during adding an external sst file. this could cause the file to be deleted while we are adding it.
This diff fix this issue

Test Plan:
unit test to reproduce the bug
existing unit tests

Reviewers: sdong, yhchiang, andrewkr

Reviewed By: andrewkr

Subscribers: dhruba

Differential Revision: https://reviews.facebook.net/D54627
---
 db/db_impl.cc | 12 +++++++++++-
 db/db_test.cc | 52 +++++++++++++++++++++++++++++++++++++++++++++++++++
 2 files changed, 63 insertions(+), 1 deletion(-)

diff --git a/db/db_impl.cc b/db/db_impl.cc
index 683006388..fc8c531c4 100644
--- a/db/db_impl.cc
+++ b/db/db_impl.cc
@@ -3587,8 +3587,16 @@ Status DBImpl::AddFile(ColumnFamilyHandle* column_family,
     return Status::InvalidArgument(
         "Non zero sequence numbers are not supported");
   }
+
   // Generate a location for the new table
-  meta.fd = FileDescriptor(versions_->NewFileNumber(), 0, file_info->file_size);
+  std::list<uint64_t>::iterator pending_outputs_inserted_elem;
+  {
+    InstrumentedMutexLock l(&mutex_);
+    pending_outputs_inserted_elem = CaptureCurrentFileNumberInPendingOutputs();
+    meta.fd =
+        FileDescriptor(versions_->NewFileNumber(), 0, file_info->file_size);
+  }
+
   std::string db_fname = TableFileName(
       db_options_.db_paths, meta.fd.GetNumber(), meta.fd.GetPathId());
 
@@ -3601,6 +3609,7 @@ Status DBImpl::AddFile(ColumnFamilyHandle* column_family,
   } else {
     status = CopyFile(env_, file_info->file_path, db_fname, 0);
   }
+  TEST_SYNC_POINT("DBImpl::AddFile:FileCopied");
   if (!status.ok()) {
     return status;
   }
@@ -3664,6 +3673,7 @@ Status DBImpl::AddFile(ColumnFamilyHandle* column_family,
       delete InstallSuperVersionAndScheduleWork(cfd, nullptr,
                                                 mutable_cf_options);
     }
+    ReleaseFileNumberFromPendingOutputs(pending_outputs_inserted_elem);
   }
 
   if (!status.ok()) {
diff --git a/db/db_test.cc b/db/db_test.cc
index 08e4edd93..de9bf8209 100644
--- a/db/db_test.cc
+++ b/db/db_test.cc
@@ -9409,6 +9409,58 @@ TEST_F(DBTest, AddExternalSstFile) {
                          kSkipFIFOCompaction));
 }
 
+// This test reporduce a bug that can happen in some cases if the DB started
+// purging obsolete files when we are adding an external sst file.
+// This situation may result in deleting the file while it's being added.
+TEST_F(DBTest, AddExternalSstFilePurgeObsoleteFilesBug) {
+  std::string sst_files_folder = test::TmpDir(env_) + "/sst_files/";
+  env_->CreateDir(sst_files_folder);
+  Options options = CurrentOptions();
+  options.env = env_;
+  const ImmutableCFOptions ioptions(options);
+
+  SstFileWriter sst_file_writer(EnvOptions(), ioptions, options.comparator);
+
+  // file1.sst (0 => 500)
+  std::string sst_file_path = sst_files_folder + "file1.sst";
+  Status s = sst_file_writer.Open(sst_file_path);
+  ASSERT_OK(s);
+  for (int i = 0; i < 500; i++) {
+    std::string k = Key(i);
+    s = sst_file_writer.Add(k, k + "_val");
+    ASSERT_OK(s);
+  }
+
+  ExternalSstFileInfo sst_file_info;
+  s = sst_file_writer.Finish(&sst_file_info);
+  ASSERT_OK(s);
+
+  options.delete_obsolete_files_period_micros = 0;
+  options.disable_auto_compactions = true;
+  DestroyAndReopen(options);
+
+  rocksdb::SyncPoint::GetInstance()->SetCallBack(
+      "DBImpl::AddFile:FileCopied", [&](void* arg) {
+        ASSERT_OK(Put("aaa", "bbb"));
+        ASSERT_OK(Flush());
+        ASSERT_OK(Put("aaa", "xxx"));
+        ASSERT_OK(Flush());
+        db_->CompactRange(CompactRangeOptions(), nullptr, nullptr);
+      });
+  rocksdb::SyncPoint::GetInstance()->EnableProcessing();
+
+  s = db_->AddFile(sst_file_path);
+  ASSERT_OK(s);
+
+  for (int i = 0; i < 500; i++) {
+    std::string k = Key(i);
+    std::string v = k + "_val";
+    ASSERT_EQ(Get(k), v);
+  }
+
+  rocksdb::SyncPoint::GetInstance()->DisableProcessing();
+}
+
 TEST_F(DBTest, AddExternalSstFileNoCopy) {
   std::string sst_files_folder = test::TmpDir(env_) + "/sst_files/";
   env_->CreateDir(sst_files_folder);