Rely on PurgeObsoleteFiles Only for Options file clean up when remote compaction is enabled (#13139)

Summary:
In PR https://github.com/facebook/rocksdb/issues/13074 , we added a logic to prevent stale OPTIONS file from getting deleted by `PurgeObsoleteFiles()` if the OPTIONS file is being referenced by any of the scheduled the remote compactions.

`PurgeObsoleteFiles()` was not the only place that we were cleaning up the old OPTIONS file. We've been also directly cleaning up the old OPTIONS file as part of `SetOptions()`: `RenameTempFileToOptionsFile()` -> `DeleteObsoleteOptionsFiles()` unless FileDeletion is disabled.

This was not caught by the UnitTest because we always preserve the last two OPTIONS file. A single call of `SetOptions()` was not enough to surface this issue in the previous PR.

To keep things simple, we are just skipping the old OPTIONS file clean up in `RenameTempFileToOptionsFile()` if remote compaction is enabled. We let `PurgeObsoleteFiles()` clean up the old options file later after the compaction is done.

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

Test Plan:
Updated UnitTest to reproduce the scenario. It's now passing with the fix.

```
./compaction_service_test --gtest_filter="*PreservedOptionsRemoteCompaction*"
```

Reviewed By: cbi42

Differential Revision: D65974726

Pulled By: jaykorean

fbshipit-source-id: 1907e8450d2ccbb42a93084f275e666648ef5b8c

db/compaction/compaction_service_test.cc

@@ -483,21 +483,29 @@ TEST_F(CompactionServiceTest, PreservedOptionsRemoteCompaction) {
     ASSERT_OK(Flush());
   }
 
-  bool is_primary_called = false;
-  // This will be called twice. One from primary and one from remote.
-  // Try changing the option when called from remote. Otherwise, the new option
-  // will be used
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->LoadDependency(
+      {{"CompactionServiceTest::OptionsFileChanged",
+        "DBImplSecondary::OpenAndCompact::BeforeLoadingOptions:1"}});
+
   ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack(
-      "DBImpl::BackgroundCompaction:NonTrivial:BeforeRun", [&](void* /*arg*/) {
-        if (!is_primary_called) {
-          is_primary_called = true;
-          return;
-        }
-        // Change the option right before the compaction run
+      "DBImplSecondary::OpenAndCompact::BeforeLoadingOptions:0",
+      [&](void* arg) {
+        auto options_file_number = static_cast<uint64_t*>(arg);
+        // Change the option twice before the compaction run
         ASSERT_OK(dbfull()->SetOptions(
             {{"level0_file_num_compaction_trigger", "4"}}));
         ASSERT_EQ(4, dbfull()->GetOptions().level0_file_num_compaction_trigger);
-        dbfull()->TEST_DeleteObsoleteFiles();
+        ASSERT_TRUE(dbfull()->versions_->options_file_number() >
+                    *options_file_number);
+
+        // Change the option twice before the compaction run
+        ASSERT_OK(dbfull()->SetOptions(
+            {{"level0_file_num_compaction_trigger", "5"}}));
+        ASSERT_EQ(5, dbfull()->GetOptions().level0_file_num_compaction_trigger);
+        ASSERT_TRUE(dbfull()->versions_->options_file_number() >
+                    *options_file_number);
+
+        TEST_SYNC_POINT("CompactionServiceTest::OptionsFileChanged");
       });
 
   ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack(

db/db_impl/db_impl.cc

@@ -5514,7 +5514,8 @@ Status DBImpl::WriteOptionsFile(const WriteOptions& write_options,
                                    file_name, fs_.get());
 
   if (s.ok()) {
-    s = RenameTempFileToOptionsFile(file_name);
+    s = RenameTempFileToOptionsFile(file_name,
+                                    db_options.compaction_service != nullptr);
   }
 
   if (!s.ok() && GetEnv()->FileExists(file_name).ok()) {
@@ -5591,7 +5592,8 @@ Status DBImpl::DeleteObsoleteOptionsFiles() {
   return Status::OK();
 }
 
-Status DBImpl::RenameTempFileToOptionsFile(const std::string& file_name) {
+Status DBImpl::RenameTempFileToOptionsFile(const std::string& file_name,
+                                           bool is_remote_compaction_enabled) {
   Status s;
 
   uint64_t options_file_number = versions_->NewFileNumber();
@@ -5635,7 +5637,7 @@ Status DBImpl::RenameTempFileToOptionsFile(const std::string& file_name) {
       my_disable_delete_obsolete_files = disable_delete_obsolete_files_;
     }
 
-    if (!my_disable_delete_obsolete_files) {
+    if (!my_disable_delete_obsolete_files && !is_remote_compaction_enabled) {
       // TODO: Should we check for errors here?
       DeleteObsoleteOptionsFiles().PermitUncheckedError();
     }

db/db_impl/db_impl.h

@@ -1469,7 +1469,8 @@ class DBImpl : public DB {
   // The following two functions can only be called when:
   // 1. WriteThread::Writer::EnterUnbatched() is used.
   // 2. db_mutex is NOT held
-  Status RenameTempFileToOptionsFile(const std::string& file_name);
+  Status RenameTempFileToOptionsFile(const std::string& file_name,
+                                     bool is_remote_compaction_enabled);
   Status DeleteObsoleteOptionsFiles();
 
   void NotifyOnManualFlushScheduled(autovector<ColumnFamilyData*> cfds,

db/db_impl/db_impl_secondary.cc

@@ -958,6 +958,10 @@ Status DB::OpenAndCompact(
   config_options.env = override_options.env;
   std::vector<ColumnFamilyDescriptor> all_column_families;
 
+  TEST_SYNC_POINT_CALLBACK(
+      "DBImplSecondary::OpenAndCompact::BeforeLoadingOptions:0",
+      &compaction_input.options_file_number);
+  TEST_SYNC_POINT("DBImplSecondary::OpenAndCompact::BeforeLoadingOptions:1");
   std::string options_file_name =
       OptionsFileName(name, compaction_input.options_file_number);
 

options/options_helper.cc

@@ -201,6 +201,7 @@ void BuildDBOptions(const ImmutableDBOptions& immutable_db_options,
   options.metadata_write_temperature =
       immutable_db_options.metadata_write_temperature;
   options.wal_write_temperature = immutable_db_options.wal_write_temperature;
+  options.compaction_service = immutable_db_options.compaction_service;
 }
 
 ColumnFamilyOptions BuildColumnFamilyOptions(