Version and history for 9.7.4

Summary:
An earlier change (https://github.com/facebook/rocksdb/commit/b34cef57b798520791312f2f40681c4d12d5d33c) removed apparently unused functionality where an obsolete blob file number is passed for removal from TableCache, which manages SST files. This was actually relying on broken/fragile abstractions wherein TableCache and BlobFileCache share the same Cache and using the TableCache interface to manipulate blob file caching. No unit test was actually checking for removal of obsolete blob files from the cache (which is somewhat tricky to check and a second order correctness requirement).

Here we fix the leak and add a DEBUG+ASAN-only check in DB::Close() that no obsolete files are lingering in the table/blob file cache.

Fixes https://github.com/facebook/rocksdb/issues/13066

Important follow-up (FIXME): The added check discovered some apparent cases of leaked (into table_cache) SST file readers that would stick around until DB::Close(). Need to enable that check, diagnose, and fix.

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

Test Plan:
added a check that is called during DB::Close in ASAN builds (to minimize paying the cost in all unit tests). Without the fix, the check failed in at least these tests:

```
db_blob_basic_test DBBlobBasicTest.DynamicallyWarmCacheDuringFlush
db_blob_compaction_test DBBlobCompactionTest.CompactionReadaheadMerge
db_blob_compaction_test DBBlobCompactionTest.MergeBlobWithBase
db_blob_compaction_test DBBlobCompactionTest.CompactionDoNotFillCache
db_blob_compaction_test DBBlobCompactionTest.SkipUntilFilter
db_blob_compaction_test DBBlobCompactionTest.CompactionFilter
db_blob_compaction_test DBBlobCompactionTest.CompactionReadaheadFilter
db_blob_compaction_test DBBlobCompactionTest.CompactionReadaheadGarbageCollection
```

Reviewed By: ltamasi

Differential Revision: D65296123

Pulled By: pdillinger

fbshipit-source-id: 2276d76482beb2c75c9010bc1bec070bb23a24c0

HISTORY.md

@@ -1,6 +1,14 @@
 # Rocksdb Change Log
 > NOTE: Entries for next release do not go here. Follow instructions in `unreleased_history/README.txt`
 
+## 9.7.4 (10/31/2024)
+### Bug Fixes
+* Fix a leak of obsolete blob files left open until DB::Close(). This bug was introduced in version 9.4.0.
+
+## 9.7.3 (10/16/2024)
+### Behavior Changes
+* OPTIONS file to be loaded by remote worker is now preserved so that it does not get purged by the primary host. A similar technique as how we are preserving new SST files from getting purged is used for this. min_options_file_numbers_ is tracked like pending_outputs_ is tracked.
+
 ## 9.7.2 (10/08/2024)
 ### Bug Fixes
 * Fix a bug for surfacing write unix time: `Iterator::GetProperty("rocksdb.iterator.write-time")` for non-L0 files.

db/blob/blob_file_cache.cc

@@ -42,6 +42,7 @@ Status BlobFileCache::GetBlobFileReader(
   assert(blob_file_reader);
   assert(blob_file_reader->IsEmpty());
 
+  // NOTE: sharing same Cache with table_cache
   const Slice key = GetSliceForKey(&blob_file_number);
 
   assert(cache_);
@@ -98,4 +99,13 @@ Status BlobFileCache::GetBlobFileReader(
   return Status::OK();
 }
 
+void BlobFileCache::Evict(uint64_t blob_file_number) {
+  // NOTE: sharing same Cache with table_cache
+  const Slice key = GetSliceForKey(&blob_file_number);
+
+  assert(cache_);
+
+  cache_.get()->Erase(key);
+}
+
 }  // namespace ROCKSDB_NAMESPACE

db/blob/blob_file_cache.h

@@ -36,6 +36,15 @@ class BlobFileCache {
                            uint64_t blob_file_number,
                            CacheHandleGuard<BlobFileReader>* blob_file_reader);
 
+  // Called when a blob file is obsolete to ensure it is removed from the cache
+  // to avoid effectively leaking the open file and assicated memory
+  void Evict(uint64_t blob_file_number);
+
+  // Used to identify cache entries for blob files (not normally useful)
+  static const Cache::CacheItemHelper* GetHelper() {
+    return CacheInterface::GetBasicHelper();
+  }
+
  private:
   using CacheInterface =
       BasicTypedCacheInterface<BlobFileReader, CacheEntryRole::kMisc>;

db/column_family.h

@@ -401,6 +401,7 @@ class ColumnFamilyData {
                          SequenceNumber earliest_seq);
 
   TableCache* table_cache() const { return table_cache_.get(); }
+  BlobFileCache* blob_file_cache() const { return blob_file_cache_.get(); }
   BlobSource* blob_source() const { return blob_source_.get(); }
 
   // See documentation in compaction_picker.h

db/compaction/compaction_job.cc

@@ -806,6 +806,12 @@ Status CompactionJob::Run() {
     }
   }
 
+  // Before the compaction starts, is_remote_compaction was set to true if
+  // compaction_service is set. We now know whether each sub_compaction was
+  // done remotely or not. Reset is_remote_compaction back to false and allow
+  // AggregateCompactionStats() to set the right value.
+  compaction_job_stats_->is_remote_compaction = false;
+
   // Finish up all bookkeeping to unify the subcompaction results.
   compact_->AggregateCompactionStats(compaction_stats_, *compaction_job_stats_);
   uint64_t num_input_range_del = 0;
@@ -1084,6 +1090,7 @@ void CompactionJob::ProcessKeyValueCompaction(SubcompactionState* sub_compact) {
     }
     // fallback to local compaction
     assert(comp_status == CompactionServiceJobStatus::kUseLocal);
+    sub_compact->compaction_job_stats.is_remote_compaction = false;
   }
 
   uint64_t prev_cpu_micros = db_options_.clock->CPUMicros();

db/compaction/compaction_job.h

@@ -209,6 +209,8 @@ class CompactionJob {
   // Returns true iff compaction_stats_.stats.num_input_records and
   // num_input_range_del are calculated successfully.
   bool UpdateCompactionStats(uint64_t* num_input_range_del = nullptr);
+  virtual void UpdateCompactionJobStats(
+      const InternalStats::CompactionStats& stats) const;
   void LogCompaction();
   virtual void RecordCompactionIOStats();
   void CleanupCompaction();
@@ -279,8 +281,7 @@ class CompactionJob {
                                   bool* compaction_released);
   Status OpenCompactionOutputFile(SubcompactionState* sub_compact,
                                   CompactionOutputs& outputs);
-  void UpdateCompactionJobStats(
-      const InternalStats::CompactionStats& stats) const;
+
   void RecordDroppedKeys(const CompactionIterationStats& c_iter_stats,
                          CompactionJobStats* compaction_job_stats = nullptr);
 
@@ -396,6 +397,8 @@ struct CompactionServiceInput {
   bool has_end = false;
   std::string end;
 
+  uint64_t options_file_number;
+
   // serialization interface to read and write the object
   static Status Read(const std::string& data_str, CompactionServiceInput* obj);
   Status Write(std::string* output);
@@ -413,19 +416,22 @@ struct CompactionServiceOutputFile {
   SequenceNumber largest_seqno;
   std::string smallest_internal_key;
   std::string largest_internal_key;
-  uint64_t oldest_ancester_time;
-  uint64_t file_creation_time;
-  uint64_t epoch_number;
+  uint64_t oldest_ancester_time = kUnknownOldestAncesterTime;
+  uint64_t file_creation_time = kUnknownFileCreationTime;
+  uint64_t epoch_number = kUnknownEpochNumber;
+  std::string file_checksum = kUnknownFileChecksum;
+  std::string file_checksum_func_name = kUnknownFileChecksumFuncName;
   uint64_t paranoid_hash;
   bool marked_for_compaction;
-  UniqueId64x2 unique_id;
+  UniqueId64x2 unique_id{};
 
   CompactionServiceOutputFile() = default;
   CompactionServiceOutputFile(
       const std::string& name, SequenceNumber smallest, SequenceNumber largest,
       std::string _smallest_internal_key, std::string _largest_internal_key,
       uint64_t _oldest_ancester_time, uint64_t _file_creation_time,
-      uint64_t _epoch_number, uint64_t _paranoid_hash,
+      uint64_t _epoch_number, const std::string& _file_checksum,
+      const std::string& _file_checksum_func_name, uint64_t _paranoid_hash,
       bool _marked_for_compaction, UniqueId64x2 _unique_id)
       : file_name(name),
         smallest_seqno(smallest),
@@ -435,6 +441,8 @@ struct CompactionServiceOutputFile {
         oldest_ancester_time(_oldest_ancester_time),
         file_creation_time(_file_creation_time),
         epoch_number(_epoch_number),
+        file_checksum(_file_checksum),
+        file_checksum_func_name(_file_checksum_func_name),
         paranoid_hash(_paranoid_hash),
         marked_for_compaction(_marked_for_compaction),
         unique_id(std::move(_unique_id)) {}
@@ -451,9 +459,6 @@ struct CompactionServiceResult {
   // location of the output files
   std::string output_path;
 
-  // some statistics about the compaction
-  uint64_t num_output_records = 0;
-  uint64_t total_bytes = 0;
   uint64_t bytes_read = 0;
   uint64_t bytes_written = 0;
   CompactionJobStats stats;
@@ -499,6 +504,9 @@ class CompactionServiceCompactionJob : private CompactionJob {
  protected:
   void RecordCompactionIOStats() override;
 
+  void UpdateCompactionJobStats(
+      const InternalStats::CompactionStats& stats) const override;
+
  private:
   // Get table file name in output_path
   std::string GetTableFileName(uint64_t file_number) override;

db/compaction/compaction_job_test.cc

@@ -1656,20 +1656,30 @@ TEST_F(CompactionJobTest, ResultSerialization) {
   };
   result.status =
       status_list.at(rnd.Uniform(static_cast<int>(status_list.size())));
+
+  std::string file_checksum = rnd.RandomBinaryString(rnd.Uniform(kStrMaxLen));
+  std::string file_checksum_func_name = "MyAwesomeChecksumGenerator";
   while (!rnd.OneIn(10)) {
     UniqueId64x2 id{rnd64.Uniform(UINT64_MAX), rnd64.Uniform(UINT64_MAX)};
     result.output_files.emplace_back(
-        rnd.RandomString(rnd.Uniform(kStrMaxLen)), rnd64.Uniform(UINT64_MAX),
-        rnd64.Uniform(UINT64_MAX),
-        rnd.RandomBinaryString(rnd.Uniform(kStrMaxLen)),
-        rnd.RandomBinaryString(rnd.Uniform(kStrMaxLen)),
-        rnd64.Uniform(UINT64_MAX), rnd64.Uniform(UINT64_MAX),
-        rnd64.Uniform(UINT64_MAX), rnd64.Uniform(UINT64_MAX), rnd.OneIn(2), id);
+        rnd.RandomString(rnd.Uniform(kStrMaxLen)) /* file_name */,
+        rnd64.Uniform(UINT64_MAX) /* smallest_seqno */,
+        rnd64.Uniform(UINT64_MAX) /* largest_seqno */,
+        rnd.RandomBinaryString(
+            rnd.Uniform(kStrMaxLen)) /* smallest_internal_key */,
+        rnd.RandomBinaryString(
+            rnd.Uniform(kStrMaxLen)) /* largest_internal_key */,
+        rnd64.Uniform(UINT64_MAX) /* oldest_ancester_time */,
+        rnd64.Uniform(UINT64_MAX) /* file_creation_time */,
+        rnd64.Uniform(UINT64_MAX) /* epoch_number */,
+        file_checksum /* file_checksum */,
+        file_checksum_func_name /* file_checksum_func_name */,
+        rnd64.Uniform(UINT64_MAX) /* paranoid_hash */,
+        rnd.OneIn(2) /* marked_for_compaction */, id);
   }
   result.output_level = rnd.Uniform(10);
   result.output_path = rnd.RandomString(rnd.Uniform(kStrMaxLen));
-  result.num_output_records = rnd64.Uniform(UINT64_MAX);
-  result.total_bytes = rnd64.Uniform(UINT64_MAX);
+  result.stats.num_output_records = rnd64.Uniform(UINT64_MAX);
   result.bytes_read = 123;
   result.bytes_written = rnd64.Uniform(UINT64_MAX);
   result.stats.elapsed_micros = rnd64.Uniform(UINT64_MAX);
@@ -1700,6 +1710,10 @@ TEST_F(CompactionJobTest, ResultSerialization) {
     ASSERT_FALSE(deserialized_tmp.TEST_Equals(&result, &mismatch));
     ASSERT_EQ(mismatch, "output_files.unique_id");
     deserialized_tmp.status.PermitUncheckedError();
+
+    ASSERT_EQ(deserialized_tmp.output_files[0].file_checksum, file_checksum);
+    ASSERT_EQ(deserialized_tmp.output_files[0].file_checksum_func_name,
+              file_checksum_func_name);
   }
 
   // Test unknown field

db/compaction/compaction_outputs.h

@@ -62,8 +62,9 @@ class CompactionOutputs {
   }
 
   // TODO: Remove it when remote compaction support tiered compaction
-  void SetTotalBytes(uint64_t bytes) { stats_.bytes_written += bytes; }
+  void AddBytesWritten(uint64_t bytes) { stats_.bytes_written += bytes; }
   void SetNumOutputRecords(uint64_t num) { stats_.num_output_records = num; }
+  void SetNumOutputFiles(uint64_t num) { stats_.num_output_files = num; }
 
   // TODO: Move the BlobDB builder into CompactionOutputs
   const std::vector<BlobFileAddition>& GetBlobFileAdditions() const {

db/compaction/compaction_service_job.cc

@@ -48,6 +48,14 @@ CompactionJob::ProcessKeyValueCompactionWithCompactionService(
   compaction_input.has_end = sub_compact->end.has_value();
   compaction_input.end =
       compaction_input.has_end ? sub_compact->end->ToString() : "";
+  compaction_input.options_file_number =
+      sub_compact->compaction->input_version()
+          ->version_set()
+          ->options_file_number();
+
+  TEST_SYNC_POINT_CALLBACK(
+      "CompactionServiceJob::ProcessKeyValueCompactionWithCompactionService",
+      &compaction_input);
 
   std::string compaction_input_binary;
   Status s = compaction_input.Write(&compaction_input_binary);
@@ -195,6 +203,8 @@ CompactionJob::ProcessKeyValueCompactionWithCompactionService(
     meta.oldest_ancester_time = file.oldest_ancester_time;
     meta.file_creation_time = file.file_creation_time;
     meta.epoch_number = file.epoch_number;
+    meta.file_checksum = file.file_checksum;
+    meta.file_checksum_func_name = file.file_checksum_func_name;
     meta.marked_for_compaction = file.marked_for_compaction;
     meta.unique_id = file.unique_id;
 
@@ -205,8 +215,10 @@ CompactionJob::ProcessKeyValueCompactionWithCompactionService(
   }
   sub_compact->compaction_job_stats = compaction_result.stats;
   sub_compact->Current().SetNumOutputRecords(
-      compaction_result.num_output_records);
-  sub_compact->Current().SetTotalBytes(compaction_result.total_bytes);
+      compaction_result.stats.num_output_records);
+  sub_compact->Current().SetNumOutputFiles(
+      compaction_result.stats.num_output_files);
+  sub_compact->Current().AddBytesWritten(compaction_result.bytes_written);
   RecordTick(stats_, REMOTE_COMPACT_READ_BYTES, compaction_result.bytes_read);
   RecordTick(stats_, REMOTE_COMPACT_WRITE_BYTES,
              compaction_result.bytes_written);
@@ -226,6 +238,18 @@ void CompactionServiceCompactionJob::RecordCompactionIOStats() {
   CompactionJob::RecordCompactionIOStats();
 }
 
+void CompactionServiceCompactionJob::UpdateCompactionJobStats(
+    const InternalStats::CompactionStats& stats) const {
+  compaction_job_stats_->elapsed_micros = stats.micros;
+
+  // output information only in remote compaction
+  compaction_job_stats_->total_output_bytes = stats.bytes_written;
+  compaction_job_stats_->total_output_bytes_blob = stats.bytes_written_blob;
+  compaction_job_stats_->num_output_records = stats.num_output_records;
+  compaction_job_stats_->num_output_files = stats.num_output_files;
+  compaction_job_stats_->num_output_files_blob = stats.num_output_files_blob;
+}
+
 CompactionServiceCompactionJob::CompactionServiceCompactionJob(
     int job_id, Compaction* compaction, const ImmutableDBOptions& db_options,
     const MutableDBOptions& mutable_db_options, const FileOptions& file_options,
@@ -280,6 +304,9 @@ Status CompactionServiceCompactionJob::Run() {
 
   log_buffer_->FlushBufferToLog();
   LogCompaction();
+
+  compaction_result_->stats.Reset();
+
   const uint64_t start_micros = db_options_.clock->NowMicros();
   c->GetOrInitInputTableProperties();
 
@@ -320,39 +347,47 @@ Status CompactionServiceCompactionJob::Run() {
   if (status.ok()) {
     status = io_s;
   }
-  if (status.ok()) {
-    // TODO: Add verify_table()
-  }
-
-  // Finish up all book-keeping to unify the subcompaction results
-  compact_->AggregateCompactionStats(compaction_stats_, *compaction_job_stats_);
-  UpdateCompactionStats();
-  RecordCompactionIOStats();
 
   LogFlush(db_options_.info_log);
   compact_->status = status;
   compact_->status.PermitUncheckedError();
 
-  // Build compaction result
+  // Build Compaction Job Stats
+
+  // 1. Aggregate CompactionOutputStats into Internal Compaction Stats
+  // (compaction_stats_) and aggregate Compaction Job Stats
+  // (compaction_job_stats_) from the sub compactions
+  compact_->AggregateCompactionStats(compaction_stats_, *compaction_job_stats_);
+
+  // 2. Update the Output information in the Compaction Job Stats with
+  // aggregated Internal Compaction Stats.
+  UpdateCompactionJobStats(compaction_stats_.stats);
+
+  // 3. Set fields that are not propagated as part of aggregations above
+  compaction_result_->stats.is_manual_compaction = c->is_manual_compaction();
+  compaction_result_->stats.is_full_compaction = c->is_full_compaction();
+  compaction_result_->stats.is_remote_compaction = true;
+
+  // 4. Update IO Stats that are not part of the aggregations above (bytes_read,
+  // bytes_written)
+  RecordCompactionIOStats();
+
+  // Build Output
   compaction_result_->output_level = compact_->compaction->output_level();
   compaction_result_->output_path = output_path_;
-  compaction_result_->stats.is_remote_compaction = true;
   for (const auto& output_file : sub_compact->GetOutputs()) {
     auto& meta = output_file.meta;
     compaction_result_->output_files.emplace_back(
         MakeTableFileName(meta.fd.GetNumber()), meta.fd.smallest_seqno,
         meta.fd.largest_seqno, meta.smallest.Encode().ToString(),
         meta.largest.Encode().ToString(), meta.oldest_ancester_time,
-        meta.file_creation_time, meta.epoch_number,
-        output_file.validator.GetHash(), meta.marked_for_compaction,
-        meta.unique_id);
+        meta.file_creation_time, meta.epoch_number, meta.file_checksum,
+        meta.file_checksum_func_name, output_file.validator.GetHash(),
+        meta.marked_for_compaction, meta.unique_id);
   }
-  InternalStats::CompactionStatsFull compaction_stats;
-  sub_compact->AggregateCompactionStats(compaction_stats);
-  compaction_result_->num_output_records =
-      compaction_stats.stats.num_output_records;
-  compaction_result_->total_bytes = compaction_stats.TotalBytesWritten();
 
+  TEST_SYNC_POINT_CALLBACK("CompactionServiceCompactionJob::Run:0",
+                           &compaction_result_);
   return status;
 }
 
@@ -435,6 +470,10 @@ static std::unordered_map<std::string, OptionTypeInfo> cs_input_type_info = {
     {"end",
      {offsetof(struct CompactionServiceInput, end), OptionType::kEncodedString,
       OptionVerificationType::kNormal, OptionTypeFlags::kNone}},
+    {"options_file_number",
+     {offsetof(struct CompactionServiceInput, options_file_number),
+      OptionType::kUInt64T, OptionVerificationType::kNormal,
+      OptionTypeFlags::kNone}},
 };
 
 static std::unordered_map<std::string, OptionTypeInfo>
@@ -471,6 +510,14 @@ static std::unordered_map<std::string, OptionTypeInfo>
          {offsetof(struct CompactionServiceOutputFile, epoch_number),
           OptionType::kUInt64T, OptionVerificationType::kNormal,
           OptionTypeFlags::kNone}},
+        {"file_checksum",
+         {offsetof(struct CompactionServiceOutputFile, file_checksum),
+          OptionType::kEncodedString, OptionVerificationType::kNormal,
+          OptionTypeFlags::kNone}},
+        {"file_checksum_func_name",
+         {offsetof(struct CompactionServiceOutputFile, file_checksum_func_name),
+          OptionType::kEncodedString, OptionVerificationType::kNormal,
+          OptionTypeFlags::kNone}},
         {"paranoid_hash",
          {offsetof(struct CompactionServiceOutputFile, paranoid_hash),
           OptionType::kUInt64T, OptionVerificationType::kNormal,
@@ -703,14 +750,6 @@ static std::unordered_map<std::string, OptionTypeInfo> cs_result_type_info = {
      {offsetof(struct CompactionServiceResult, output_path),
       OptionType::kEncodedString, OptionVerificationType::kNormal,
       OptionTypeFlags::kNone}},
-    {"num_output_records",
-     {offsetof(struct CompactionServiceResult, num_output_records),
-      OptionType::kUInt64T, OptionVerificationType::kNormal,
-      OptionTypeFlags::kNone}},
-    {"total_bytes",
-     {offsetof(struct CompactionServiceResult, total_bytes),
-      OptionType::kUInt64T, OptionVerificationType::kNormal,
-      OptionTypeFlags::kNone}},
     {"bytes_read",
      {offsetof(struct CompactionServiceResult, bytes_read),
       OptionType::kUInt64T, OptionVerificationType::kNormal,

db/compaction/compaction_service_test.cc

@@ -3,9 +3,9 @@
 //  COPYING file in the root directory) and Apache 2.0 License
 //  (found in the LICENSE.Apache file in the root directory).
 
-
 #include "db/db_test_util.h"
 #include "port/stack_trace.h"
+#include "rocksdb/utilities/options_util.h"
 #include "table/unique_id_impl.h"
 
 namespace ROCKSDB_NAMESPACE {
@@ -349,7 +349,27 @@ TEST_F(CompactionServiceTest, BasicCompactions) {
   } else {
     ASSERT_OK(result.status);
   }
+  ASSERT_GE(result.stats.elapsed_micros, 1);
+  ASSERT_GE(result.stats.cpu_micros, 1);
+
+  ASSERT_EQ(20, result.stats.num_output_records);
+  ASSERT_EQ(result.output_files.size(), result.stats.num_output_files);
+
+  uint64_t total_size = 0;
+  for (auto output_file : result.output_files) {
+    std::string file_name = result.output_path + "/" + output_file.file_name;
+
+    uint64_t file_size = 0;
+    ASSERT_OK(options.env->GetFileSize(file_name, &file_size));
+    ASSERT_GT(file_size, 0);
+    total_size += file_size;
+  }
+  ASSERT_EQ(total_size, result.stats.total_output_bytes);
+
   ASSERT_TRUE(result.stats.is_remote_compaction);
+  ASSERT_TRUE(result.stats.is_manual_compaction);
+  ASSERT_FALSE(result.stats.is_full_compaction);
+
   Close();
 }
 
@@ -396,6 +416,501 @@ TEST_F(CompactionServiceTest, ManualCompaction) {
   ASSERT_TRUE(result.stats.is_remote_compaction);
 }
 
+TEST_F(CompactionServiceTest, PreservedOptionsLocalCompaction) {
+  Options options = CurrentOptions();
+  options.level0_file_num_compaction_trigger = 2;
+  options.disable_auto_compactions = true;
+  DestroyAndReopen(options);
+
+  Random rnd(301);
+  for (auto i = 0; i < 2; ++i) {
+    for (auto j = 0; j < 10; ++j) {
+      ASSERT_OK(
+          Put("foo" + std::to_string(i * 10 + j), rnd.RandomString(1024)));
+    }
+    ASSERT_OK(Flush());
+  }
+
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack(
+      "CompactionJob::ProcessKeyValueCompaction()::Processing", [&](void* arg) {
+        auto compaction = static_cast<Compaction*>(arg);
+        std::string options_file_name = OptionsFileName(
+            dbname_,
+            compaction->input_version()->version_set()->options_file_number());
+
+        // Change option twice to make sure the very first OPTIONS file gets
+        // purged
+        ASSERT_OK(dbfull()->SetOptions(
+            {{"level0_file_num_compaction_trigger", "4"}}));
+        ASSERT_EQ(4, dbfull()->GetOptions().level0_file_num_compaction_trigger);
+        ASSERT_OK(dbfull()->SetOptions(
+            {{"level0_file_num_compaction_trigger", "6"}}));
+        ASSERT_EQ(6, dbfull()->GetOptions().level0_file_num_compaction_trigger);
+        dbfull()->TEST_DeleteObsoleteFiles();
+
+        // For non-remote compactions, OPTIONS file can be deleted while
+        // using option at the start of the compaction
+        Status s = env_->FileExists(options_file_name);
+        ASSERT_NOK(s);
+        ASSERT_TRUE(s.IsNotFound());
+        // Should be old value
+        ASSERT_EQ(2, compaction->mutable_cf_options()
+                         ->level0_file_num_compaction_trigger);
+        ASSERT_TRUE(dbfull()->min_options_file_numbers_.empty());
+      });
+
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->EnableProcessing();
+  Status s = dbfull()->CompactRange(CompactRangeOptions(), nullptr, nullptr);
+  ASSERT_TRUE(s.ok());
+}
+
+TEST_F(CompactionServiceTest, PreservedOptionsRemoteCompaction) {
+  // For non-remote compaction do not preserve options file
+  Options options = CurrentOptions();
+  options.level0_file_num_compaction_trigger = 2;
+  options.disable_auto_compactions = true;
+  ReopenWithCompactionService(&options);
+  GenerateTestData();
+
+  auto my_cs = GetCompactionService();
+
+  Random rnd(301);
+  for (auto i = 0; i < 2; ++i) {
+    for (auto j = 0; j < 10; ++j) {
+      ASSERT_OK(
+          Put("foo" + std::to_string(i * 10 + j), rnd.RandomString(1024)));
+    }
+    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()->SetCallBack(
+      "DBImpl::BackgroundCompaction:NonTrivial:BeforeRun", [&](void* /*arg*/) {
+        if (!is_primary_called) {
+          is_primary_called = true;
+          return;
+        }
+        // Change the option right 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();
+      });
+
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack(
+      "CompactionServiceJob::ProcessKeyValueCompactionWithCompactionService",
+      [&](void* arg) {
+        auto input = static_cast<CompactionServiceInput*>(arg);
+        std::string options_file_name =
+            OptionsFileName(dbname_, input->options_file_number);
+
+        ASSERT_OK(env_->FileExists(options_file_name));
+        ASSERT_FALSE(dbfull()->min_options_file_numbers_.empty());
+        ASSERT_EQ(dbfull()->min_options_file_numbers_.front(),
+                  input->options_file_number);
+
+        DBOptions db_options;
+        ConfigOptions config_options;
+        std::vector<ColumnFamilyDescriptor> all_column_families;
+        config_options.env = env_;
+        ASSERT_OK(LoadOptionsFromFile(config_options, options_file_name,
+                                      &db_options, &all_column_families));
+        bool has_cf = false;
+        for (auto& cf : all_column_families) {
+          if (cf.name == input->cf_name) {
+            // Should be old value
+            ASSERT_EQ(2, cf.options.level0_file_num_compaction_trigger);
+            has_cf = true;
+          }
+        }
+        ASSERT_TRUE(has_cf);
+      });
+
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack(
+      "CompactionJob::ProcessKeyValueCompaction()::Processing", [&](void* arg) {
+        auto compaction = static_cast<Compaction*>(arg);
+        ASSERT_EQ(2, compaction->mutable_cf_options()
+                         ->level0_file_num_compaction_trigger);
+      });
+
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->EnableProcessing();
+  Status s = dbfull()->CompactRange(CompactRangeOptions(), nullptr, nullptr);
+  ASSERT_TRUE(s.ok());
+
+  CompactionServiceResult result;
+  my_cs->GetResult(&result);
+  ASSERT_OK(result.status);
+  ASSERT_TRUE(result.stats.is_manual_compaction);
+  ASSERT_TRUE(result.stats.is_remote_compaction);
+}
+
+class EventVerifier : public EventListener {
+ public:
+  explicit EventVerifier(uint64_t expected_num_input_records,
+                         size_t expected_num_input_files,
+                         uint64_t expected_num_output_records,
+                         size_t expected_num_output_files,
+                         const std::string& expected_smallest_output_key_prefix,
+                         const std::string& expected_largest_output_key_prefix,
+                         bool expected_is_remote_compaction_on_begin,
+                         bool expected_is_remote_compaction_on_complete)
+      : expected_num_input_records_(expected_num_input_records),
+        expected_num_input_files_(expected_num_input_files),
+        expected_num_output_records_(expected_num_output_records),
+        expected_num_output_files_(expected_num_output_files),
+        expected_smallest_output_key_prefix_(
+            expected_smallest_output_key_prefix),
+        expected_largest_output_key_prefix_(expected_largest_output_key_prefix),
+        expected_is_remote_compaction_on_begin_(
+            expected_is_remote_compaction_on_begin),
+        expected_is_remote_compaction_on_complete_(
+            expected_is_remote_compaction_on_complete) {}
+  void OnCompactionBegin(DB* /*db*/, const CompactionJobInfo& ci) override {
+    ASSERT_EQ(expected_num_input_files_, ci.input_files.size());
+    ASSERT_EQ(expected_num_input_files_, ci.input_file_infos.size());
+    ASSERT_EQ(expected_is_remote_compaction_on_begin_,
+              ci.stats.is_remote_compaction);
+    ASSERT_TRUE(ci.stats.is_manual_compaction);
+    ASSERT_FALSE(ci.stats.is_full_compaction);
+  }
+  void OnCompactionCompleted(DB* /*db*/, const CompactionJobInfo& ci) override {
+    ASSERT_GT(ci.stats.elapsed_micros, 0);
+    ASSERT_GT(ci.stats.cpu_micros, 0);
+    ASSERT_EQ(expected_num_input_records_, ci.stats.num_input_records);
+    ASSERT_EQ(expected_num_input_files_, ci.stats.num_input_files);
+    ASSERT_EQ(expected_num_output_records_, ci.stats.num_output_records);
+    ASSERT_EQ(expected_num_output_files_, ci.stats.num_output_files);
+    ASSERT_EQ(expected_smallest_output_key_prefix_,
+              ci.stats.smallest_output_key_prefix);
+    ASSERT_EQ(expected_largest_output_key_prefix_,
+              ci.stats.largest_output_key_prefix);
+    ASSERT_GT(ci.stats.total_input_bytes, 0);
+    ASSERT_GT(ci.stats.total_output_bytes, 0);
+    ASSERT_EQ(ci.stats.num_input_records,
+              ci.stats.num_output_records + ci.stats.num_records_replaced);
+    ASSERT_EQ(expected_is_remote_compaction_on_complete_,
+              ci.stats.is_remote_compaction);
+    ASSERT_TRUE(ci.stats.is_manual_compaction);
+    ASSERT_FALSE(ci.stats.is_full_compaction);
+  }
+
+ private:
+  uint64_t expected_num_input_records_;
+  size_t expected_num_input_files_;
+  uint64_t expected_num_output_records_;
+  size_t expected_num_output_files_;
+  std::string expected_smallest_output_key_prefix_;
+  std::string expected_largest_output_key_prefix_;
+  bool expected_is_remote_compaction_on_begin_;
+  bool expected_is_remote_compaction_on_complete_;
+};
+
+TEST_F(CompactionServiceTest, VerifyStats) {
+  Options options = CurrentOptions();
+  options.disable_auto_compactions = true;
+  auto event_verifier = std::make_shared<EventVerifier>(
+      30 /* expected_num_input_records */, 3 /* expected_num_input_files */,
+      20 /* expected_num_output_records */, 1 /* expected_num_output_files */,
+      "key00000" /* expected_smallest_output_key_prefix */,
+      "key00001" /* expected_largest_output_key_prefix */,
+      true /* expected_is_remote_compaction_on_begin */,
+      true /* expected_is_remote_compaction_on_complete */);
+  options.listeners.push_back(event_verifier);
+  ReopenWithCompactionService(&options);
+  GenerateTestData();
+
+  auto my_cs = GetCompactionService();
+
+  std::string start_str = Key(0);
+  std::string end_str = Key(1);
+  Slice start(start_str);
+  Slice end(end_str);
+  uint64_t comp_num = my_cs->GetCompactionNum();
+  ASSERT_OK(db_->CompactRange(CompactRangeOptions(), &start, &end));
+  ASSERT_GE(my_cs->GetCompactionNum(), comp_num + 1);
+  VerifyTestData();
+
+  CompactionServiceResult result;
+  my_cs->GetResult(&result);
+  ASSERT_OK(result.status);
+  ASSERT_TRUE(result.stats.is_manual_compaction);
+  ASSERT_TRUE(result.stats.is_remote_compaction);
+}
+
+TEST_F(CompactionServiceTest, VerifyStatsLocalFallback) {
+  Options options = CurrentOptions();
+  options.disable_auto_compactions = true;
+  auto event_verifier = std::make_shared<EventVerifier>(
+      30 /* expected_num_input_records */, 3 /* expected_num_input_files */,
+      20 /* expected_num_output_records */, 1 /* expected_num_output_files */,
+      "key00000" /* expected_smallest_output_key_prefix */,
+      "key00001" /* expected_largest_output_key_prefix */,
+      true /* expected_is_remote_compaction_on_begin */,
+      false /* expected_is_remote_compaction_on_complete */);
+  options.listeners.push_back(event_verifier);
+  ReopenWithCompactionService(&options);
+  GenerateTestData();
+
+  auto my_cs = GetCompactionService();
+  my_cs->OverrideStartStatus(CompactionServiceJobStatus::kUseLocal);
+
+  std::string start_str = Key(0);
+  std::string end_str = Key(1);
+  Slice start(start_str);
+  Slice end(end_str);
+  uint64_t comp_num = my_cs->GetCompactionNum();
+  ASSERT_OK(db_->CompactRange(CompactRangeOptions(), &start, &end));
+  // Remote Compaction did not happen
+  ASSERT_EQ(my_cs->GetCompactionNum(), comp_num);
+  VerifyTestData();
+}
+
+TEST_F(CompactionServiceTest, CorruptedOutput) {
+  Options options = CurrentOptions();
+  options.disable_auto_compactions = true;
+  ReopenWithCompactionService(&options);
+  GenerateTestData();
+
+  auto my_cs = GetCompactionService();
+
+  std::string start_str = Key(15);
+  std::string end_str = Key(45);
+  Slice start(start_str);
+  Slice end(end_str);
+  uint64_t comp_num = my_cs->GetCompactionNum();
+
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack(
+      "CompactionServiceCompactionJob::Run:0", [&](void* arg) {
+        CompactionServiceResult* compaction_result =
+            *(static_cast<CompactionServiceResult**>(arg));
+        ASSERT_TRUE(compaction_result != nullptr &&
+                    !compaction_result->output_files.empty());
+        // Corrupt files here
+        for (const auto& output_file : compaction_result->output_files) {
+          std::string file_name =
+              compaction_result->output_path + "/" + output_file.file_name;
+
+          uint64_t file_size = 0;
+          Status s = options.env->GetFileSize(file_name, &file_size);
+          ASSERT_OK(s);
+          ASSERT_GT(file_size, 0);
+
+          ASSERT_OK(test::CorruptFile(env_, file_name, 0,
+                                      static_cast<int>(file_size),
+                                      true /* verifyChecksum */));
+        }
+      });
+  SyncPoint::GetInstance()->EnableProcessing();
+
+  // CompactRange() should fail
+  Status s = db_->CompactRange(CompactRangeOptions(), &start, &end);
+  ASSERT_NOK(s);
+  ASSERT_TRUE(s.IsCorruption());
+
+  ASSERT_GE(my_cs->GetCompactionNum(), comp_num + 1);
+
+  SyncPoint::GetInstance()->DisableProcessing();
+  SyncPoint::GetInstance()->ClearAllCallBacks();
+
+  // On the worker side, the compaction is considered success
+  // Verification is done on the primary side
+  CompactionServiceResult result;
+  my_cs->GetResult(&result);
+  ASSERT_OK(result.status);
+  ASSERT_TRUE(result.stats.is_manual_compaction);
+  ASSERT_TRUE(result.stats.is_remote_compaction);
+}
+
+TEST_F(CompactionServiceTest, CorruptedOutputParanoidFileCheck) {
+  for (bool paranoid_file_check_enabled : {false, true}) {
+    SCOPED_TRACE("paranoid_file_check_enabled=" +
+                 std::to_string(paranoid_file_check_enabled));
+
+    Options options = CurrentOptions();
+    Destroy(options);
+    options.disable_auto_compactions = true;
+    options.paranoid_file_checks = paranoid_file_check_enabled;
+    ReopenWithCompactionService(&options);
+    GenerateTestData();
+
+    auto my_cs = GetCompactionService();
+
+    std::string start_str = Key(15);
+    std::string end_str = Key(45);
+    Slice start(start_str);
+    Slice end(end_str);
+    uint64_t comp_num = my_cs->GetCompactionNum();
+
+    ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack(
+        "CompactionServiceCompactionJob::Run:0", [&](void* arg) {
+          CompactionServiceResult* compaction_result =
+              *(static_cast<CompactionServiceResult**>(arg));
+          ASSERT_TRUE(compaction_result != nullptr &&
+                      !compaction_result->output_files.empty());
+          // Corrupt files here
+          for (const auto& output_file : compaction_result->output_files) {
+            std::string file_name =
+                compaction_result->output_path + "/" + output_file.file_name;
+
+            // Corrupt very small range of bytes. This corruption is so small
+            // that this isn't caught by default light-weight check
+            ASSERT_OK(test::CorruptFile(env_, file_name, 0, 1,
+                                        false /* verifyChecksum */));
+          }
+        });
+    SyncPoint::GetInstance()->EnableProcessing();
+
+    Status s = db_->CompactRange(CompactRangeOptions(), &start, &end);
+    if (paranoid_file_check_enabled) {
+      ASSERT_NOK(s);
+      ASSERT_EQ(Status::Corruption("Paranoid checksums do not match"), s);
+    } else {
+      // CompactRange() goes through if paranoid file check is not enabled
+      ASSERT_OK(s);
+    }
+
+    ASSERT_GE(my_cs->GetCompactionNum(), comp_num + 1);
+
+    SyncPoint::GetInstance()->DisableProcessing();
+    SyncPoint::GetInstance()->ClearAllCallBacks();
+
+    // On the worker side, the compaction is considered success
+    // Verification is done on the primary side
+    CompactionServiceResult result;
+    my_cs->GetResult(&result);
+    ASSERT_OK(result.status);
+    ASSERT_TRUE(result.stats.is_manual_compaction);
+    ASSERT_TRUE(result.stats.is_remote_compaction);
+  }
+}
+
+TEST_F(CompactionServiceTest, TruncatedOutput) {
+  Options options = CurrentOptions();
+  options.disable_auto_compactions = true;
+  ReopenWithCompactionService(&options);
+  GenerateTestData();
+
+  auto my_cs = GetCompactionService();
+
+  std::string start_str = Key(15);
+  std::string end_str = Key(45);
+  Slice start(start_str);
+  Slice end(end_str);
+  uint64_t comp_num = my_cs->GetCompactionNum();
+
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack(
+      "CompactionServiceCompactionJob::Run:0", [&](void* arg) {
+        CompactionServiceResult* compaction_result =
+            *(static_cast<CompactionServiceResult**>(arg));
+        ASSERT_TRUE(compaction_result != nullptr &&
+                    !compaction_result->output_files.empty());
+        // Truncate files here
+        for (const auto& output_file : compaction_result->output_files) {
+          std::string file_name =
+              compaction_result->output_path + "/" + output_file.file_name;
+
+          uint64_t file_size = 0;
+          Status s = options.env->GetFileSize(file_name, &file_size);
+          ASSERT_OK(s);
+          ASSERT_GT(file_size, 0);
+
+          ASSERT_OK(test::TruncateFile(env_, file_name, file_size / 2));
+        }
+      });
+  SyncPoint::GetInstance()->EnableProcessing();
+
+  // CompactRange() should fail
+  Status s = db_->CompactRange(CompactRangeOptions(), &start, &end);
+  ASSERT_NOK(s);
+  ASSERT_TRUE(s.IsCorruption());
+
+  ASSERT_GE(my_cs->GetCompactionNum(), comp_num + 1);
+
+  SyncPoint::GetInstance()->DisableProcessing();
+  SyncPoint::GetInstance()->ClearAllCallBacks();
+
+  // On the worker side, the compaction is considered success
+  // Verification is done on the primary side
+  CompactionServiceResult result;
+  my_cs->GetResult(&result);
+  ASSERT_OK(result.status);
+  ASSERT_TRUE(result.stats.is_manual_compaction);
+  ASSERT_TRUE(result.stats.is_remote_compaction);
+}
+
+TEST_F(CompactionServiceTest, CustomFileChecksum) {
+  Options options = CurrentOptions();
+  options.file_checksum_gen_factory = GetFileChecksumGenCrc32cFactory();
+  ReopenWithCompactionService(&options);
+  GenerateTestData();
+
+  auto my_cs = GetCompactionService();
+
+  std::string start_str = Key(15);
+  std::string end_str = Key(45);
+  Slice start(start_str);
+  Slice end(end_str);
+  uint64_t comp_num = my_cs->GetCompactionNum();
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack(
+      "CompactionServiceCompactionJob::Run:0", [&](void* arg) {
+        CompactionServiceResult* compaction_result =
+            *(static_cast<CompactionServiceResult**>(arg));
+        ASSERT_TRUE(compaction_result != nullptr &&
+                    !compaction_result->output_files.empty());
+        // Validate Checksum files here
+        for (const auto& output_file : compaction_result->output_files) {
+          std::string file_name =
+              compaction_result->output_path + "/" + output_file.file_name;
+
+          FileChecksumGenContext gen_context;
+          gen_context.file_name = file_name;
+          std::unique_ptr<FileChecksumGenerator> file_checksum_gen =
+              options.file_checksum_gen_factory->CreateFileChecksumGenerator(
+                  gen_context);
+
+          std::unique_ptr<SequentialFile> file_reader;
+          uint64_t file_size = 0;
+          Status s = options.env->GetFileSize(file_name, &file_size);
+          ASSERT_OK(s);
+          ASSERT_GT(file_size, 0);
+
+          s = options.env->NewSequentialFile(file_name, &file_reader,
+                                             EnvOptions());
+          ASSERT_OK(s);
+
+          Slice result;
+          std::unique_ptr<char[]> scratch(new char[file_size]);
+          s = file_reader->Read(file_size, &result, scratch.get());
+          ASSERT_OK(s);
+
+          file_checksum_gen->Update(scratch.get(), result.size());
+          file_checksum_gen->Finalize();
+
+          // Verify actual checksum and the func name
+          ASSERT_EQ(file_checksum_gen->Name(),
+                    output_file.file_checksum_func_name);
+          ASSERT_EQ(file_checksum_gen->GetChecksum(),
+                    output_file.file_checksum);
+        }
+      });
+  SyncPoint::GetInstance()->EnableProcessing();
+
+  ASSERT_OK(db_->CompactRange(CompactRangeOptions(), &start, &end));
+  ASSERT_GE(my_cs->GetCompactionNum(), comp_num + 1);
+
+  SyncPoint::GetInstance()->DisableProcessing();
+  SyncPoint::GetInstance()->ClearAllCallBacks();
+
+  CompactionServiceResult result;
+  my_cs->GetResult(&result);
+  ASSERT_OK(result.status);
+  ASSERT_TRUE(result.stats.is_manual_compaction);
+  ASSERT_TRUE(result.stats.is_remote_compaction);
+}
+
 TEST_F(CompactionServiceTest, CancelCompactionOnRemoteSide) {
   Options options = CurrentOptions();
   options.disable_auto_compactions = true;

db/compaction/compaction_state.cc

@@ -39,7 +39,7 @@ void CompactionState::AggregateCompactionStats(
     InternalStats::CompactionStatsFull& compaction_stats,
     CompactionJobStats& compaction_job_stats) {
   for (const auto& sc : sub_compact_states) {
-    sc.AggregateCompactionStats(compaction_stats);
+    sc.AggregateCompactionOutputStats(compaction_stats);
     compaction_job_stats.Add(sc.compaction_job_stats);
   }
 }

db/compaction/subcompaction_state.cc

@@ -13,7 +13,7 @@
 #include "rocksdb/sst_partitioner.h"
 
 namespace ROCKSDB_NAMESPACE {
-void SubcompactionState::AggregateCompactionStats(
+void SubcompactionState::AggregateCompactionOutputStats(
     InternalStats::CompactionStatsFull& compaction_stats) const {
   compaction_stats.stats.Add(compaction_outputs_.stats_);
   if (HasPenultimateLevelOutputs()) {

db/compaction/subcompaction_state.h

@@ -179,7 +179,7 @@ class SubcompactionState {
 
   void Cleanup(Cache* cache);
 
-  void AggregateCompactionStats(
+  void AggregateCompactionOutputStats(
       InternalStats::CompactionStatsFull& compaction_stats) const;
 
   CompactionOutputs& Current() const {

db/db_impl/db_impl.cc

@@ -659,8 +659,9 @@ Status DBImpl::CloseHelper() {
   // We need to release them before the block cache is destroyed. The block
   // cache may be destroyed inside versions_.reset(), when column family data
   // list is destroyed, so leaving handles in table cache after
-  // versions_.reset() may cause issues.
-  // Here we clean all unreferenced handles in table cache.
+  // versions_.reset() may cause issues. Here we clean all unreferenced handles
+  // in table cache, and (for certain builds/conditions) assert that no obsolete
+  // files are hanging around unreferenced (leak) in the table/blob file cache.
   // Now we assume all user queries have finished, so only version set itself
   // can possibly hold the blocks from block cache. After releasing unreferenced
   // handles here, only handles held by version set left and inside
@@ -668,6 +669,9 @@ Status DBImpl::CloseHelper() {
   // time a handle is released, we erase it from the cache too. By doing that,
   // we can guarantee that after versions_.reset(), table cache is empty
   // so the cache can be safely destroyed.
+#ifndef NDEBUG
+  TEST_VerifyNoObsoleteFilesCached(/*db_mutex_already_held=*/true);
+#endif  // !NDEBUG
   table_cache_->EraseUnRefEntries();
 
   for (auto& txn_entry : recovered_transactions_) {
@@ -3227,6 +3231,8 @@ Status DBImpl::MultiGetImpl(
       s = Status::Aborted();
       break;
     }
+    // This could be a long-running operation
+    ROCKSDB_THREAD_YIELD_HOOK();
   }
 
   // Post processing (decrement reference counts and record statistics)
@@ -4779,6 +4785,24 @@ void DBImpl::ReleaseFileNumberFromPendingOutputs(
   }
 }
 
+std::list<uint64_t>::iterator DBImpl::CaptureOptionsFileNumber() {
+  // We need to remember the iterator of our insert, because after the
+  // compaction is done, we need to remove that element from
+  // min_options_file_numbers_.
+  min_options_file_numbers_.push_back(versions_->options_file_number());
+  auto min_options_file_numbers_inserted_elem = min_options_file_numbers_.end();
+  --min_options_file_numbers_inserted_elem;
+  return min_options_file_numbers_inserted_elem;
+}
+
+void DBImpl::ReleaseOptionsFileNumber(
+    std::unique_ptr<std::list<uint64_t>::iterator>& v) {
+  if (v.get() != nullptr) {
+    min_options_file_numbers_.erase(*v.get());
+    v.reset();
+  }
+}
+
 Status DBImpl::GetUpdatesSince(
     SequenceNumber seq, std::unique_ptr<TransactionLogIterator>* iter,
     const TransactionLogIterator::ReadOptions& read_options) {

db/db_impl/db_impl.h

@@ -853,6 +853,8 @@ class DBImpl : public DB {
 
   uint64_t GetObsoleteSstFilesSize();
 
+  uint64_t MinOptionsFileNumberToKeep();
+
   // Returns the list of live files in 'live' and the list
   // of all files in the filesystem in 'candidate_files'.
   // If force == false and the last call was less than
@@ -1239,9 +1241,14 @@ class DBImpl : public DB {
   static Status TEST_ValidateOptions(const DBOptions& db_options) {
     return ValidateOptions(db_options);
   }
-
 #endif  // NDEBUG
 
+  // In certain configurations, verify that the table/blob file cache only
+  // contains entries for live files, to check for effective leaks of open
+  // files. This can only be called when purging of obsolete files has
+  // "settled," such as during parts of DB Close().
+  void TEST_VerifyNoObsoleteFilesCached(bool db_mutex_already_held) const;
+
   // persist stats to column family "_persistent_stats"
   void PersistStats();
 
@@ -1694,6 +1701,8 @@ class DBImpl : public DB {
   friend class XFTransactionWriteHandler;
   friend class DBBlobIndexTest;
   friend class WriteUnpreparedTransactionTest_RecoveryTest_Test;
+  friend class CompactionServiceTest_PreservedOptionsLocalCompaction_Test;
+  friend class CompactionServiceTest_PreservedOptionsRemoteCompaction_Test;
 #endif
 
   struct CompactionState;
@@ -1965,6 +1974,12 @@ class DBImpl : public DB {
   void ReleaseFileNumberFromPendingOutputs(
       std::unique_ptr<std::list<uint64_t>::iterator>& v);
 
+  // Similar to pending_outputs, preserve OPTIONS file. Used for remote
+  // compaction.
+  std::list<uint64_t>::iterator CaptureOptionsFileNumber();
+  void ReleaseOptionsFileNumber(
+      std::unique_ptr<std::list<uint64_t>::iterator>& v);
+
   // Sets bg error if there is an error writing to WAL.
   IOStatus SyncClosedWals(const WriteOptions& write_options,
                           JobContext* job_context, VersionEdit* synced_wals,
@@ -2755,6 +2770,11 @@ class DBImpl : public DB {
   // State is protected with db mutex.
   std::list<uint64_t> pending_outputs_;
 
+  // Similar to pending_outputs_, FindObsoleteFiles()/PurgeObsoleteFiles() never
+  // deletes any OPTIONS file that has number bigger than any of the file number
+  // in min_options_file_numbers_.
+  std::list<uint64_t> min_options_file_numbers_;
+
   // flush_queue_ and compaction_queue_ hold column families that we need to
   // flush and compact, respectively.
   // A column family is inserted into flush_queue_ when it satisfies condition

db/db_impl/db_impl_compaction_flush.cc

@@ -1561,6 +1561,12 @@ Status DBImpl::CompactFilesImpl(
 
   compaction_job.Prepare();
 
+  std::unique_ptr<std::list<uint64_t>::iterator> min_options_file_number_elem;
+  if (immutable_db_options().compaction_service != nullptr) {
+    min_options_file_number_elem.reset(
+        new std::list<uint64_t>::iterator(CaptureOptionsFileNumber()));
+  }
+
   mutex_.Unlock();
   TEST_SYNC_POINT("CompactFilesImpl:0");
   TEST_SYNC_POINT("CompactFilesImpl:1");
@@ -1570,6 +1576,10 @@ Status DBImpl::CompactFilesImpl(
   TEST_SYNC_POINT("CompactFilesImpl:3");
   mutex_.Lock();
 
+  if (immutable_db_options().compaction_service != nullptr) {
+    ReleaseOptionsFileNumber(min_options_file_number_elem);
+  }
+
   bool compaction_released = false;
   Status status =
       compaction_job.Install(*c->mutable_cf_options(), &compaction_released);
@@ -3541,6 +3551,14 @@ Status DBImpl::BackgroundCompaction(bool* made_progress,
       is_manual && manual_compaction->disallow_trivial_move;
 
   CompactionJobStats compaction_job_stats;
+  // Set is_remote_compaction to true on CompactionBegin Event if
+  // compaction_service is set except for trivial moves. We do not know whether
+  // remote compaction will actually be successfully scheduled, or fall back to
+  // local at this time. CompactionCompleted event will tell the truth where
+  // the compaction actually happened.
+  compaction_job_stats.is_remote_compaction =
+      immutable_db_options().compaction_service != nullptr;
+
   Status status;
   if (!error_handler_.IsBGWorkStopped()) {
     if (shutting_down_.load(std::memory_order_acquire)) {
@@ -3766,6 +3784,8 @@ Status DBImpl::BackgroundCompaction(bool* made_progress,
     ThreadStatusUtil::SetThreadOperation(ThreadStatus::OP_COMPACTION);
 
     compaction_job_stats.num_input_files = c->num_input_files(0);
+    // Trivial moves do not get compacted remotely
+    compaction_job_stats.is_remote_compaction = false;
 
     NotifyOnCompactionBegin(c->column_family_data(), c.get(), status,
                             compaction_job_stats, job_context->job_id);
@@ -3901,6 +3921,12 @@ Status DBImpl::BackgroundCompaction(bool* made_progress,
         &bg_bottom_compaction_scheduled_);
     compaction_job.Prepare();
 
+    std::unique_ptr<std::list<uint64_t>::iterator> min_options_file_number_elem;
+    if (immutable_db_options().compaction_service != nullptr) {
+      min_options_file_number_elem.reset(
+          new std::list<uint64_t>::iterator(CaptureOptionsFileNumber()));
+    }
+
     NotifyOnCompactionBegin(c->column_family_data(), c.get(), status,
                             compaction_job_stats, job_context->job_id);
     mutex_.Unlock();
@@ -3910,6 +3936,11 @@ Status DBImpl::BackgroundCompaction(bool* made_progress,
     compaction_job.Run().PermitUncheckedError();
     TEST_SYNC_POINT("DBImpl::BackgroundCompaction:NonTrivial:AfterRun");
     mutex_.Lock();
+
+    if (immutable_db_options().compaction_service != nullptr) {
+      ReleaseOptionsFileNumber(min_options_file_number_elem);
+    }
+
     status =
         compaction_job.Install(*c->mutable_cf_options(), &compaction_released);
     io_s = compaction_job.io_status();

db/db_impl/db_impl_debug.cc

@@ -9,6 +9,7 @@
 
 #ifndef NDEBUG
 
+#include "db/blob/blob_file_cache.h"
 #include "db/column_family.h"
 #include "db/db_impl/db_impl.h"
 #include "db/error_handler.h"
@@ -328,5 +329,49 @@ size_t DBImpl::TEST_EstimateInMemoryStatsHistorySize() const {
   InstrumentedMutexLock l(&const_cast<DBImpl*>(this)->stats_history_mutex_);
   return EstimateInMemoryStatsHistorySize();
 }
+
+void DBImpl::TEST_VerifyNoObsoleteFilesCached(
+    bool db_mutex_already_held) const {
+  // This check is somewhat expensive and obscure to make a part of every
+  // unit test in every build variety. Thus, we only enable it for ASAN builds.
+  if (!kMustFreeHeapAllocations) {
+    return;
+  }
+
+  std::optional<InstrumentedMutexLock> l;
+  if (db_mutex_already_held) {
+    mutex_.AssertHeld();
+  } else {
+    l.emplace(&mutex_);
+  }
+
+  std::vector<uint64_t> live_files;
+  for (auto cfd : *versions_->GetColumnFamilySet()) {
+    if (cfd->IsDropped()) {
+      continue;
+    }
+    // Sneakily add both SST and blob files to the same list
+    cfd->current()->AddLiveFiles(&live_files, &live_files);
+  }
+  std::sort(live_files.begin(), live_files.end());
+
+  auto fn = [&live_files](const Slice& key, Cache::ObjectPtr, size_t,
+                          const Cache::CacheItemHelper* helper) {
+    if (helper != BlobFileCache::GetHelper()) {
+      // Skip non-blob files for now
+      // FIXME: diagnose and fix the leaks of obsolete SST files revealed in
+      // unit tests.
+      return;
+    }
+    // See TableCache and BlobFileCache
+    assert(key.size() == sizeof(uint64_t));
+    uint64_t file_number;
+    GetUnaligned(reinterpret_cast<const uint64_t*>(key.data()), &file_number);
+    // Assert file is in sorted live_files
+    assert(
+        std::binary_search(live_files.begin(), live_files.end(), file_number));
+  };
+  table_cache_->ApplyToAllEntries(fn, {});
+}
 }  // namespace ROCKSDB_NAMESPACE
 #endif  // NDEBUG

db/db_impl/db_impl_files.cc

@@ -43,6 +43,14 @@ uint64_t DBImpl::GetObsoleteSstFilesSize() {
   return versions_->GetObsoleteSstFilesSize();
 }
 
+uint64_t DBImpl::MinOptionsFileNumberToKeep() {
+  mutex_.AssertHeld();
+  if (!min_options_file_numbers_.empty()) {
+    return *min_options_file_numbers_.begin();
+  }
+  return std::numeric_limits<uint64_t>::max();
+}
+
 Status DBImpl::DisableFileDeletions() {
   Status s;
   int my_disable_delete_obsolete_files;
@@ -147,6 +155,7 @@ void DBImpl::FindObsoleteFiles(JobContext* job_context, bool force,
   // here but later find newer generated unfinalized files while scanning.
   job_context->min_pending_output = MinObsoleteSstNumberToKeep();
   job_context->files_to_quarantine = error_handler_.GetFilesToQuarantine();
+  job_context->min_options_file_number = MinOptionsFileNumberToKeep();
 
   // Get obsolete files.  This function will also update the list of
   // pending files in VersionSet().
@@ -498,7 +507,7 @@ void DBImpl::PurgeObsoleteFiles(JobContext& state, bool schedule_only) {
                                 dbname_);
 
   // File numbers of most recent two OPTIONS file in candidate_files (found in
-  // previos FindObsoleteFiles(full_scan=true))
+  // previous FindObsoleteFiles(full_scan=true))
   // At this point, there must not be any duplicate file numbers in
   // candidate_files.
   uint64_t optsfile_num1 = std::numeric_limits<uint64_t>::min();
@@ -519,6 +528,11 @@ void DBImpl::PurgeObsoleteFiles(JobContext& state, bool schedule_only) {
     }
   }
 
+  // For remote compactions, we need to keep OPTIONS file that may get
+  // referenced by the remote worker
+
+  optsfile_num2 = std::min(optsfile_num2, state.min_options_file_number);
+
   // Close WALs before trying to delete them.
   for (const auto w : state.logs_to_free) {
     // TODO: maybe check the return value of Close.

db/db_impl/db_impl_secondary.cc

@@ -951,21 +951,19 @@ Status DB::OpenAndCompact(
     return s;
   }
 
-  // 2. Load the options from latest OPTIONS file
+  // 2. Load the options
   DBOptions db_options;
   ConfigOptions config_options;
   config_options.env = override_options.env;
   std::vector<ColumnFamilyDescriptor> all_column_families;
-  s = LoadLatestOptions(config_options, name, &db_options,
-                        &all_column_families);
-  // In a very rare scenario, loading options may fail if the options changed by
-  // the primary host at the same time. Just retry once for now.
-  if (!s.ok()) {
-    s = LoadLatestOptions(config_options, name, &db_options,
+
+  std::string options_file_name =
+      OptionsFileName(name, compaction_input.options_file_number);
+
+  s = LoadOptionsFromFile(config_options, options_file_name, &db_options,
                           &all_column_families);
-    if (!s.ok()) {
-      return s;
-    }
+  if (!s.ok()) {
+    return s;
   }
 
   // 3. Override pointer configurations in DBOptions with

db/db_iter.cc

@@ -540,6 +540,8 @@ bool DBIter::FindNextUserEntryInternal(bool skipping_saved_key,
     } else {
       iter_.Next();
     }
+    // This could be a long-running operation due to tombstones, etc.
+    ROCKSDB_THREAD_YIELD_HOOK();
   } while (iter_.Valid());
 
   valid_ = false;

db/db_secondary_test.cc

@@ -244,7 +244,7 @@ TEST_F(DBSecondaryTest, SimpleInternalCompaction) {
   ASSERT_EQ(largest.user_key().ToString(), "foo");
   ASSERT_EQ(result.output_level, 1);
   ASSERT_EQ(result.output_path, this->secondary_path_);
-  ASSERT_EQ(result.num_output_records, 2);
+  ASSERT_EQ(result.stats.num_output_records, 2);
   ASSERT_GT(result.bytes_written, 0);
   ASSERT_OK(result.status);
 }

db/job_context.h

@@ -202,6 +202,10 @@ struct JobContext {
   // that corresponds to the set of files in 'live'.
   uint64_t manifest_file_number;
   uint64_t pending_manifest_file_number;
+
+  // Used for remote compaction. To prevent OPTIONS files from getting
+  // purged by PurgeObsoleteFiles() of the primary host
+  uint64_t min_options_file_number;
   uint64_t log_number;
   uint64_t prev_log_number;
 

db/table_cache.cc

@@ -164,6 +164,7 @@ Status TableCache::GetTableReader(
 }
 
 Cache::Handle* TableCache::Lookup(Cache* cache, uint64_t file_number) {
+  // NOTE: sharing same Cache with BlobFileCache
   Slice key = GetSliceForFileNumber(&file_number);
   return cache->Lookup(key);
 }
@@ -179,6 +180,7 @@ Status TableCache::FindTable(
     size_t max_file_size_for_l0_meta_pin, Temperature file_temperature) {
   PERF_TIMER_GUARD_WITH_CLOCK(find_table_nanos, ioptions_.clock);
   uint64_t number = file_meta.fd.GetNumber();
+  // NOTE: sharing same Cache with BlobFileCache
   Slice key = GetSliceForFileNumber(&number);
   *handle = cache_.Lookup(key);
   TEST_SYNC_POINT_CALLBACK("TableCache::FindTable:0",

db/version_builder.cc

@@ -24,6 +24,7 @@
 #include <vector>
 
 #include "cache/cache_reservation_manager.h"
+#include "db/blob/blob_file_cache.h"
 #include "db/blob/blob_file_meta.h"
 #include "db/dbformat.h"
 #include "db/internal_stats.h"
@@ -744,12 +745,9 @@ class VersionBuilder::Rep {
       return Status::Corruption("VersionBuilder", oss.str());
     }
 
-    // Note: we use C++11 for now but in C++14, this could be done in a more
-    // elegant way using generalized lambda capture.
-    VersionSet* const vs = version_set_;
-    const ImmutableCFOptions* const ioptions = ioptions_;
-
-    auto deleter = [vs, ioptions](SharedBlobFileMetaData* shared_meta) {
+    auto deleter = [vs = version_set_, ioptions = ioptions_,
+                    bc = cfd_ ? cfd_->blob_file_cache()
+                              : nullptr](SharedBlobFileMetaData* shared_meta) {
       if (vs) {
         assert(ioptions);
         assert(!ioptions->cf_paths.empty());
@@ -758,6 +756,9 @@ class VersionBuilder::Rep {
         vs->AddObsoleteBlobFile(shared_meta->GetBlobFileNumber(),
                                 ioptions->cf_paths.front().path);
       }
+      if (bc) {
+        bc->Evict(shared_meta->GetBlobFileNumber());
+      }
 
       delete shared_meta;
     };
@@ -766,7 +767,7 @@ class VersionBuilder::Rep {
         blob_file_number, blob_file_addition.GetTotalBlobCount(),
         blob_file_addition.GetTotalBlobBytes(),
         blob_file_addition.GetChecksumMethod(),
-        blob_file_addition.GetChecksumValue(), deleter);
+        blob_file_addition.GetChecksumValue(), std::move(deleter));
 
     mutable_blob_file_metas_.emplace(
         blob_file_number, MutableBlobFileMetaData(std::move(shared_meta)));

db/version_set.h

@@ -1514,7 +1514,6 @@ class VersionSet {
   void GetLiveFilesMetaData(std::vector<LiveFileMetaData>* metadata);
 
   void AddObsoleteBlobFile(uint64_t blob_file_number, std::string path) {
-    // TODO: Erase file from BlobFileCache?
     obsolete_blob_files_.emplace_back(blob_file_number, std::move(path));
   }
 

db/write_batch.cc

@@ -502,7 +502,9 @@ Status ReadRecordFromWriteBatch(Slice* input, char* tag,
       break;
     }
     default:
-      return Status::Corruption("unknown WriteBatch tag");
+      return Status::Corruption(
+          "unknown WriteBatch tag",
+          std::to_string(static_cast<unsigned int>(*tag)));
   }
   return Status::OK();
 }
@@ -750,7 +752,9 @@ Status WriteBatchInternal::Iterate(const WriteBatch* wb,
         }
         break;
       default:
-        return Status::Corruption("unknown WriteBatch tag");
+        return Status::Corruption(
+            "unknown WriteBatch tag",
+            std::to_string(static_cast<unsigned int>(tag)));
     }
   }
   if (!s.ok()) {

include/rocksdb/version.h

@@ -13,7 +13,7 @@
 // minor or major version number planned for release.
 #define ROCKSDB_MAJOR 9
 #define ROCKSDB_MINOR 7
-#define ROCKSDB_PATCH 2
+#define ROCKSDB_PATCH 4
 
 // Do not use these. We made the mistake of declaring macros starting with
 // double underscore. Now we have to live with our choice. We'll deprecate these

port/port.h

@@ -19,3 +19,19 @@
 #elif defined(OS_WIN)
 #include "port/win/port_win.h"
 #endif
+
+#ifdef OS_LINUX
+// A temporary hook into long-running RocksDB threads to support modifying their
+// priority etc. This should become a public API hook once the requirements
+// are better understood.
+extern "C" void RocksDbThreadYield() __attribute__((__weak__));
+#define ROCKSDB_THREAD_YIELD_HOOK() \
+  {                                 \
+    if (RocksDbThreadYield) {       \
+      RocksDbThreadYield();         \
+    }                               \
+  }
+#else
+#define ROCKSDB_THREAD_YIELD_HOOK() \
+  {}
+#endif

util/compaction_job_stats_impl.cc

@@ -89,6 +89,8 @@ void CompactionJobStats::Add(const CompactionJobStats& stats) {
 
   num_single_del_fallthru += stats.num_single_del_fallthru;
   num_single_del_mismatch += stats.num_single_del_mismatch;
+
+  is_remote_compaction |= stats.is_remote_compaction;
 }