Add an option to wait for purge in WaitForCompact (#12520)

Summary:
Adding an option to wait for purge to complete in `WaitForCompact` API.

Internally, RocksDB has a way to wait for purge to complete (e.g. TEST_WaitForPurge() in db_impl_debug.cc), but there's no public API available for gracefully wait for purge to complete.

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

Test Plan:
Unit Test Added - `WaitForCompactWithWaitForPurgeOptionTest`
```
./deletefile_test -- --gtest_filter="*WaitForCompactWithWaitForPurgeOptionTest*"
```

Existing Tests
```
./db_compaction_test -- --gtest_filter="*WaitForCompactWithOption*"
```

Reviewed By: ajkr

Differential Revision: D55888283

Pulled By: jaykorean

fbshipit-source-id: cfc6d6e8657deaefab8961890b36e390095c9f65

db/db_impl/db_impl.cc

@@ -1854,6 +1854,7 @@ void DBImpl::SchedulePurge() {
 }
 
 void DBImpl::BackgroundCallPurge() {
+  TEST_SYNC_POINT("DBImpl::BackgroundCallPurge:beforeMutexLock");
   mutex_.Lock();
 
   while (!logs_to_free_queue_.empty()) {

db/db_impl/db_impl_compaction_flush.cc

@@ -4356,6 +4356,7 @@ Status DBImpl::WaitForCompact(
     }
     if ((bg_bottom_compaction_scheduled_ || bg_compaction_scheduled_ ||
          bg_flush_scheduled_ || unscheduled_compactions_ ||
+         (wait_for_compact_options.wait_for_purge && bg_purge_scheduled_) ||
          unscheduled_flushes_ || error_handler_.IsRecoveryInProgress()) &&
         (error_handler_.GetBGError().ok())) {
       if (wait_for_compact_options.timeout.count()) {
@@ -4363,6 +4364,7 @@ Status DBImpl::WaitForCompact(
           return Status::TimedOut();
         }
       } else {
+        TEST_SYNC_POINT("DBImpl::WaitForCompact:InsideLoop");
         bg_cv_.Wait();
       }
     } else if (wait_for_compact_options.close_db) {

db/deletefile_test.cc

@@ -223,6 +223,49 @@ TEST_F(DeleteFileTest, PurgeObsoleteFilesTest) {
   CheckFileTypeCounts(dbname_, 0, 1, 1);
 }
 
+TEST_F(DeleteFileTest, WaitForCompactWithWaitForPurgeOptionTest) {
+  Options options = CurrentOptions();
+  SetOptions(&options);
+  Destroy(options);
+  options.create_if_missing = true;
+  Reopen(options);
+
+  std::string first("0"), last("999999");
+  CompactRangeOptions compact_options;
+  compact_options.change_level = true;
+  compact_options.target_level = 2;
+  Slice first_slice(first), last_slice(last);
+
+  CreateTwoLevels();
+  Iterator* itr = nullptr;
+  ReadOptions read_options;
+  read_options.background_purge_on_iterator_cleanup = true;
+  itr = db_->NewIterator(read_options);
+  ASSERT_OK(itr->status());
+  ASSERT_OK(db_->CompactRange(compact_options, &first_slice, &last_slice));
+  SyncPoint::GetInstance()->LoadDependency(
+      {{"DBImpl::BGWorkPurge:start", "DeleteFileTest::WaitForPurgeTest"},
+       {"DBImpl::WaitForCompact:InsideLoop",
+        "DBImpl::BackgroundCallPurge:beforeMutexLock"}});
+  SyncPoint::GetInstance()->EnableProcessing();
+
+  delete itr;
+
+  TEST_SYNC_POINT("DeleteFileTest::WaitForPurgeTest");
+  // At this point, purge got started, but can't finish due to sync points
+  // not purged yet
+  CheckFileTypeCounts(dbname_, 0, 3, 1);
+
+  // The sync point in WaitForCompact should unblock the purge
+  WaitForCompactOptions wait_for_compact_options;
+  wait_for_compact_options.wait_for_purge = true;
+  Status s = dbfull()->WaitForCompact(wait_for_compact_options);
+  ASSERT_OK(s);
+
+  // Now files should be purged
+  CheckFileTypeCounts(dbname_, 0, 1, 1);
+}
+
 TEST_F(DeleteFileTest, BackgroundPurgeIteratorTest) {
   Options options = CurrentOptions();
   SetOptions(&options);
@@ -600,4 +643,3 @@ int main(int argc, char** argv) {
   RegisterCustomObjects(argc, argv);
   return RUN_ALL_TESTS();
 }
-

include/rocksdb/options.h

@@ -2212,6 +2212,9 @@ struct WaitForCompactOptions {
   // A boolean to flush all column families before starting to wait.
   bool flush = false;
 
+  // A boolean to wait for purge to complete
+  bool wait_for_purge = false;
+
   // A boolean to call Close() after waiting is done. By the time Close() is
   // called here, there should be no background jobs in progress and no new
   // background jobs should be added. DB may not have been closed if Close()

unreleased_history/new_features/wait_for_purge_option.md

@@ -0,0 +1,2 @@
+Add an option to `WaitForCompactOptions` - `wait_for_purge` to make `WaitForCompact()` API wait for background purge to complete
+