Update HISTORY.md and version.h for 9.5.2

Summary:
Add a couple of ticker stats for corruption retry count and successful retries. This PR also eliminates an extra read attempt when there's a checksum mismatch in a block read from the prefetch buffer.

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

Test Plan: Update existing tests

Reviewed By: jowlyzhang

Differential Revision: D61024687

Pulled By: anand1976

fbshipit-source-id: 3a08403580ab244000e0d480b7ee0f5a03d76b06

HISTORY.md

@@ -1,6 +1,36 @@
 # Rocksdb Change Log
 > NOTE: Entries for next release do not go here. Follow instructions in `unreleased_history/README.txt`
 
+## 9.5.2 (08/13/2024)
+### Bug Fixes
+* Fix a race condition in pessimistic transactions that could allow multiple transactions with the same name to be registered simultaneously, resulting in a crash or other unpredictable behavior.
+
+### Public API Changes
+* Add ticker stats to count file read retries due to checksum mismatch
+
+## 9.5.1 (08/02/2024)
+### Bug Fixes
+* *Make DestroyDB supports slow deletion when it's configured in `SstFileManager`. The slow deletion is subject to the configured `rate_bytes_per_sec`, but not subject to the `max_trash_db_ratio`.
+
+## 9.5.0 (07/19/2024)
+### Public API Changes
+* Introduced new C API function rocksdb_writebatch_iterate_cf for column family-aware iteration over the contents of a WriteBatch
+* Add support to ingest SST files generated by a DB instead of SstFileWriter. This can be enabled with experimental option `IngestExternalFileOptions::allow_db_generated_files`.
+
+### Behavior Changes
+* When calculating total log size for the `log_size_for_flush` argument in `CreateCheckpoint` API, the size of the archived log will not be included to avoid unnecessary flush
+
+### Bug Fixes
+* Fix a major bug in which an iterator using prefix filtering and SeekForPrev might miss data when the DB is using `whole_key_filtering=false` and `partition_filters=true`.
+* Fixed a bug where `OnErrorRecoveryBegin()` is not called before auto recovery starts.
+* Fixed a bug where event listener reads ErrorHandler's `bg_error_` member without holding db mutex(#12803).
+* Fixed a bug in handling MANIFEST write error that caused the latest valid MANIFEST file to get deleted, resulting in the DB being unopenable.
+* Fixed a race between error recovery due to manifest sync or write failure and external SST file ingestion. Both attempt to write a new manifest file, which causes an assertion failure.
+
+### Performance Improvements
+* Fix an issue where compactions were opening table files and reading table properties while holding db mutex_.
+* Reduce unnecessary filesystem queries and DB mutex acquires in creating backups and checkpoints.
+
 ## 9.4.0 (06/23/2024)
 ### New Features
 * Added a `CompactForTieringCollectorFactory` to auto trigger compaction for tiering use case.

db/db_impl/db_impl.cc

@@ -17,6 +17,7 @@
 #include <cstdio>
 #include <map>
 #include <memory>
+#include <optional>
 #include <set>
 #include <sstream>
 #include <stdexcept>
@@ -5240,6 +5241,14 @@ Status DestroyDB(const std::string& dbname, const Options& options,
   Env* env = soptions.env;
   std::vector<std::string> filenames;
   bool wal_in_db_path = soptions.IsWalDirSameAsDBPath();
+  auto sfm = static_cast_with_check<SstFileManagerImpl>(
+      options.sst_file_manager.get());
+  // Allocate a separate trash bucket to be used by all the to be deleted
+  // files, so we can later wait for this bucket to be empty before return.
+  std::optional<int32_t> bucket;
+  if (sfm) {
+    bucket = sfm->NewTrashBucket();
+  }
 
   // Reset the logger because it holds a handle to the
   // log file and prevents cleanup and directory removal
@@ -5251,6 +5260,7 @@ Status DestroyDB(const std::string& dbname, const Options& options,
                     /*IODebugContext*=*/nullptr)
       .PermitUncheckedError();
 
+  std::set<std::string> paths_to_delete;
   FileLock* lock;
   const std::string lockname = LockFileName(dbname);
   Status result = env->LockFile(lockname, &lock);
@@ -5267,10 +5277,9 @@ Status DestroyDB(const std::string& dbname, const Options& options,
           del = DestroyDB(path_to_delete, options);
         } else if (type == kTableFile || type == kWalFile ||
                    type == kBlobFile) {
-          del = DeleteDBFile(
-              &soptions, path_to_delete, dbname,
-              /*force_bg=*/false,
-              /*force_fg=*/(type == kWalFile) ? !wal_in_db_path : false);
+          del = DeleteUnaccountedDBFile(&soptions, path_to_delete, dbname,
+                                        /*force_bg=*/false,
+                                        /*force_fg=*/false, bucket);
         } else {
           del = env->DeleteFile(path_to_delete);
         }
@@ -5279,6 +5288,7 @@ Status DestroyDB(const std::string& dbname, const Options& options,
         }
       }
     }
+    paths_to_delete.insert(dbname);
 
     std::set<std::string> paths;
     for (const DbPath& db_path : options.db_paths) {
@@ -5300,18 +5310,19 @@ Status DestroyDB(const std::string& dbname, const Options& options,
               (type == kTableFile ||
                type == kBlobFile)) {  // Lock file will be deleted at end
             std::string file_path = path + "/" + fname;
-            Status del = DeleteDBFile(&soptions, file_path, dbname,
-                                      /*force_bg=*/false, /*force_fg=*/false);
+            Status del = DeleteUnaccountedDBFile(&soptions, file_path, dbname,
+                                                 /*force_bg=*/false,
+                                                 /*force_fg=*/false, bucket);
             if (!del.ok() && result.ok()) {
               result = del;
             }
           }
         }
-        // TODO: Should we return an error if we cannot delete the directory?
-        env->DeleteDir(path).PermitUncheckedError();
       }
     }
 
+    paths_to_delete.merge(paths);
+
     std::vector<std::string> walDirFiles;
     std::string archivedir = ArchivalDirectory(dbname);
     bool wal_dir_exists = false;
@@ -5335,46 +5346,49 @@ Status DestroyDB(const std::string& dbname, const Options& options,
       // Delete archival files.
       for (const auto& file : archiveFiles) {
         if (ParseFileName(file, &number, &type) && type == kWalFile) {
-          Status del =
-              DeleteDBFile(&soptions, archivedir + "/" + file, archivedir,
-                           /*force_bg=*/false, /*force_fg=*/!wal_in_db_path);
+          Status del = DeleteUnaccountedDBFile(
+              &soptions, archivedir + "/" + file, archivedir,
+              /*force_bg=*/false, /*force_fg=*/!wal_in_db_path, bucket);
           if (!del.ok() && result.ok()) {
             result = del;
           }
         }
       }
-      // Ignore error in case dir contains other files
-      env->DeleteDir(archivedir).PermitUncheckedError();
+      paths_to_delete.insert(archivedir);
     }
 
     // Delete log files in the WAL dir
     if (wal_dir_exists) {
       for (const auto& file : walDirFiles) {
         if (ParseFileName(file, &number, &type) && type == kWalFile) {
-          Status del =
-              DeleteDBFile(&soptions, LogFileName(soptions.wal_dir, number),
-                           soptions.wal_dir, /*force_bg=*/false,
-                           /*force_fg=*/!wal_in_db_path);
+          Status del = DeleteUnaccountedDBFile(
+              &soptions, LogFileName(soptions.wal_dir, number),
+              soptions.wal_dir, /*force_bg=*/false,
+              /*force_fg=*/!wal_in_db_path, bucket);
           if (!del.ok() && result.ok()) {
             result = del;
           }
         }
       }
-      // Ignore error in case dir contains other files
-      env->DeleteDir(soptions.wal_dir).PermitUncheckedError();
+      paths_to_delete.insert(soptions.wal_dir);
     }
 
     // Ignore error since state is already gone
     env->UnlockFile(lock).PermitUncheckedError();
     env->DeleteFile(lockname).PermitUncheckedError();
 
+    // Make sure trash files are all cleared before return.
+    if (sfm && bucket.has_value()) {
+      sfm->WaitForEmptyTrashBucket(bucket.value());
+    }
     // sst_file_manager holds a ref to the logger. Make sure the logger is
     // gone before trying to remove the directory.
     soptions.sst_file_manager.reset();
 
     // Ignore error in case dir contains other files
-    env->DeleteDir(dbname).PermitUncheckedError();
-    ;
+    for (const auto& path_to_delete : paths_to_delete) {
+      env->DeleteDir(path_to_delete).PermitUncheckedError();
+    }
   }
   return result;
 }

db/db_impl/db_impl.h

@@ -1226,6 +1226,8 @@ class DBImpl : public DB {
     return logs_.back().number;
   }
 
+  void TEST_DeleteObsoleteFiles();
+
   const std::unordered_set<uint64_t>& TEST_GetFilesGrabbedForPurge() const {
     return files_grabbed_for_purge_;
   }

db/db_impl/db_impl_debug.cc

@@ -314,6 +314,11 @@ const autovector<uint64_t>& DBImpl::TEST_GetFilesToQuarantine() const {
   return error_handler_.GetFilesToQuarantine();
 }
 
+void DBImpl::TEST_DeleteObsoleteFiles() {
+  InstrumentedMutexLock l(&mutex_);
+  DeleteObsoleteFiles();
+}
+
 size_t DBImpl::TEST_EstimateInMemoryStatsHistorySize() const {
   InstrumentedMutexLock l(&const_cast<DBImpl*>(this)->stats_history_mutex_);
   return EstimateInMemoryStatsHistorySize();

db/db_impl/db_impl_open.cc

@@ -530,6 +530,12 @@ Status DBImpl::Recover(
                            /*no_error_if_files_missing=*/false, is_retry,
                            &desc_status);
     desc_status.PermitUncheckedError();
+    if (is_retry) {
+      RecordTick(stats_, FILE_READ_CORRUPTION_RETRY_COUNT);
+      if (desc_status.ok()) {
+        RecordTick(stats_, FILE_READ_CORRUPTION_RETRY_SUCCESS_COUNT);
+      }
+    }
     if (can_retry) {
       // If we're opening for the first time and the failure is likely due to
       // a corrupt MANIFEST file (could result in either the log::Reader

db/db_io_failure_test.cc

@@ -705,6 +705,7 @@ class DBIOCorruptionTest
   DBIOCorruptionTest() : DBIOFailureTest() {
     BlockBasedTableOptions bbto;
     options_ = CurrentOptions();
+    options_.statistics = CreateDBStatistics();
 
     base_env_ = env_;
     EXPECT_NE(base_env_, nullptr);
@@ -727,6 +728,8 @@ class DBIOCorruptionTest
 
   Status ReopenDB() { return TryReopen(options_); }
 
+  Statistics* stats() { return options_.statistics.get(); }
+
  protected:
   std::unique_ptr<Env> env_guard_;
   std::shared_ptr<CorruptionFS> fs_;
@@ -749,8 +752,12 @@ TEST_P(DBIOCorruptionTest, GetReadCorruptionRetry) {
   if (std::get<2>(GetParam())) {
     ASSERT_OK(s);
     ASSERT_EQ(val, "val1");
+    ASSERT_EQ(stats()->getTickerCount(FILE_READ_CORRUPTION_RETRY_COUNT), 1);
+    ASSERT_EQ(stats()->getTickerCount(FILE_READ_CORRUPTION_RETRY_SUCCESS_COUNT),
+              1);
   } else {
     ASSERT_TRUE(s.IsCorruption());
+    ASSERT_EQ(stats()->getTickerCount(FILE_READ_CORRUPTION_RETRY_COUNT), 0);
   }
 }
 
@@ -773,8 +780,12 @@ TEST_P(DBIOCorruptionTest, IterReadCorruptionRetry) {
   }
   if (std::get<2>(GetParam())) {
     ASSERT_OK(iter->status());
+    ASSERT_EQ(stats()->getTickerCount(FILE_READ_CORRUPTION_RETRY_COUNT), 1);
+    ASSERT_EQ(stats()->getTickerCount(FILE_READ_CORRUPTION_RETRY_SUCCESS_COUNT),
+              1);
   } else {
     ASSERT_TRUE(iter->status().IsCorruption());
+    ASSERT_EQ(stats()->getTickerCount(FILE_READ_CORRUPTION_RETRY_COUNT), 0);
   }
   delete iter;
 }
@@ -799,9 +810,13 @@ TEST_P(DBIOCorruptionTest, MultiGetReadCorruptionRetry) {
   if (std::get<2>(GetParam())) {
     ASSERT_EQ(values[0].ToString(), "val1");
     ASSERT_EQ(values[1].ToString(), "val2");
+    ASSERT_EQ(stats()->getTickerCount(FILE_READ_CORRUPTION_RETRY_COUNT), 1);
+    ASSERT_EQ(stats()->getTickerCount(FILE_READ_CORRUPTION_RETRY_SUCCESS_COUNT),
+              1);
   } else {
     ASSERT_TRUE(statuses[0].IsCorruption());
     ASSERT_TRUE(statuses[1].IsCorruption());
+    ASSERT_EQ(stats()->getTickerCount(FILE_READ_CORRUPTION_RETRY_COUNT), 0);
   }
 }
 
@@ -818,6 +833,9 @@ TEST_P(DBIOCorruptionTest, CompactionReadCorruptionRetry) {
   Status s = dbfull()->CompactRange(CompactRangeOptions(), nullptr, nullptr);
   if (std::get<2>(GetParam())) {
     ASSERT_OK(s);
+    ASSERT_EQ(stats()->getTickerCount(FILE_READ_CORRUPTION_RETRY_COUNT), 1);
+    ASSERT_EQ(stats()->getTickerCount(FILE_READ_CORRUPTION_RETRY_SUCCESS_COUNT),
+              1);
 
     std::string val;
     ReadOptions ro;
@@ -826,6 +844,7 @@ TEST_P(DBIOCorruptionTest, CompactionReadCorruptionRetry) {
     ASSERT_EQ(val, "val1");
   } else {
     ASSERT_TRUE(s.IsCorruption());
+    ASSERT_EQ(stats()->getTickerCount(FILE_READ_CORRUPTION_RETRY_COUNT), 0);
   }
 }
 
@@ -838,6 +857,9 @@ TEST_P(DBIOCorruptionTest, FlushReadCorruptionRetry) {
   Status s = Flush();
   if (std::get<2>(GetParam())) {
     ASSERT_OK(s);
+    ASSERT_EQ(stats()->getTickerCount(FILE_READ_CORRUPTION_RETRY_COUNT), 1);
+    ASSERT_EQ(stats()->getTickerCount(FILE_READ_CORRUPTION_RETRY_SUCCESS_COUNT),
+              1);
 
     std::string val;
     ReadOptions ro;
@@ -846,6 +868,7 @@ TEST_P(DBIOCorruptionTest, FlushReadCorruptionRetry) {
     ASSERT_EQ(val, "val1");
   } else {
     ASSERT_NOK(s);
+    ASSERT_EQ(stats()->getTickerCount(FILE_READ_CORRUPTION_RETRY_COUNT), 0);
   }
 }
 
@@ -862,8 +885,12 @@ TEST_P(DBIOCorruptionTest, ManifestCorruptionRetry) {
 
   if (std::get<2>(GetParam())) {
     ASSERT_OK(ReopenDB());
+    ASSERT_EQ(stats()->getTickerCount(FILE_READ_CORRUPTION_RETRY_COUNT), 1);
+    ASSERT_EQ(stats()->getTickerCount(FILE_READ_CORRUPTION_RETRY_SUCCESS_COUNT),
+              1);
   } else {
     ASSERT_EQ(ReopenDB(), Status::Corruption());
+    ASSERT_EQ(stats()->getTickerCount(FILE_READ_CORRUPTION_RETRY_COUNT), 0);
   }
   SyncPoint::GetInstance()->DisableProcessing();
 }

db/db_sst_test.cc

@@ -507,6 +507,23 @@ TEST_F(DBSSTTest, DBWithSstFileManagerForBlobFiles) {
   ASSERT_EQ(files_deleted, 0);
   ASSERT_EQ(files_scheduled_to_delete, 0);
   Close();
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack(
+      "SstFileManagerImpl::ScheduleUnaccountedFileDeletion", [&](void* arg) {
+        assert(arg);
+        const std::string* const file_path =
+            static_cast<const std::string*>(arg);
+        if (EndsWith(*file_path, ".blob")) {
+          ++files_scheduled_to_delete;
+        }
+      });
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack(
+      "DeleteScheduler::OnDeleteFile", [&](void* arg) {
+        const std::string* const file_path =
+            static_cast<const std::string*>(arg);
+        if (EndsWith(*file_path, ".blob")) {
+          files_deleted++;
+        }
+      });
   ASSERT_OK(DestroyDB(dbname_, options));
   ASSERT_EQ(files_deleted, blob_files.size());
   ASSERT_EQ(files_scheduled_to_delete, blob_files.size());
@@ -649,6 +666,23 @@ TEST_F(DBSSTTest, DBWithSstFileManagerForBlobFilesWithGC) {
   }
 
   Close();
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack(
+      "SstFileManagerImpl::ScheduleUnaccountedFileDeletion", [&](void* arg) {
+        assert(arg);
+        const std::string* const file_path =
+            static_cast<const std::string*>(arg);
+        if (EndsWith(*file_path, ".blob")) {
+          ++files_scheduled_to_delete;
+        }
+      });
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack(
+      "DeleteScheduler::OnDeleteFile", [&](void* arg) {
+        const std::string* const file_path =
+            static_cast<const std::string*>(arg);
+        if (EndsWith(*file_path, ".blob")) {
+          files_deleted++;
+        }
+      });
   ASSERT_OK(DestroyDB(dbname_, options));
   sfm->WaitForEmptyTrash();
   ASSERT_EQ(files_deleted, 5);
@@ -1902,6 +1936,24 @@ TEST_F(DBSSTTest, DBWithSFMForBlobFilesAtomicFlush) {
   ASSERT_EQ(files_deleted, 1);
 
   Close();
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack(
+      "SstFileManagerImpl::ScheduleUnaccountedFileDeletion", [&](void* arg) {
+        assert(arg);
+        const std::string* const file_path =
+            static_cast<const std::string*>(arg);
+        if (EndsWith(*file_path, ".blob")) {
+          ++files_scheduled_to_delete;
+        }
+      });
+
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack(
+      "DeleteScheduler::OnDeleteFile", [&](void* arg) {
+        const std::string* const file_path =
+            static_cast<const std::string*>(arg);
+        if (EndsWith(*file_path, ".blob")) {
+          files_deleted++;
+        }
+      });
   ASSERT_OK(DestroyDB(dbname_, options));
 
   ASSERT_EQ(files_scheduled_to_delete, 4);

file/delete_scheduler.cc

@@ -31,6 +31,7 @@ DeleteScheduler::DeleteScheduler(SystemClock* clock, FileSystem* fs,
       total_trash_size_(0),
       rate_bytes_per_sec_(rate_bytes_per_sec),
       pending_files_(0),
+      next_trash_bucket_(0),
       bytes_max_delete_chunk_(bytes_max_delete_chunk),
       closing_(false),
       cv_(&mu_),
@@ -66,10 +67,8 @@ Status DeleteScheduler::DeleteFile(const std::string& file_path,
        total_trash_size_.load() > total_size * max_trash_db_ratio_.load())) {
     // Rate limiting is disabled or trash size makes up more than
     // max_trash_db_ratio_ (default 25%) of the total DB size
-    TEST_SYNC_POINT("DeleteScheduler::DeleteFile");
-    Status s = fs_->DeleteFile(file_path, IOOptions(), nullptr);
+    Status s = DeleteFileImmediately(file_path, /*accounted=*/true);
     if (s.ok()) {
-      s = sst_file_manager_->OnDeleteFile(file_path);
       ROCKS_LOG_INFO(info_log_,
                      "Deleted file %s immediately, rate_bytes_per_sec %" PRIi64
                      ", total_trash_size %" PRIu64 ", total_size %" PRIi64
@@ -77,15 +76,57 @@ Status DeleteScheduler::DeleteFile(const std::string& file_path,
                      file_path.c_str(), rate_bytes_per_sec_.load(),
                      total_trash_size_.load(), total_size,
                      max_trash_db_ratio_.load());
-      InstrumentedMutexLock l(&mu_);
-      RecordTick(stats_.get(), FILES_DELETED_IMMEDIATELY);
     }
     return s;
   }
+  return AddFileToDeletionQueue(file_path, dir_to_sync, /*bucket=*/std::nullopt,
+                                /*accounted=*/true);
+}
 
+Status DeleteScheduler::DeleteUnaccountedFile(const std::string& file_path,
+                                              const std::string& dir_to_sync,
+                                              const bool force_bg,
+                                              std::optional<int32_t> bucket) {
+  uint64_t num_hard_links = 1;
+  fs_->NumFileLinks(file_path, IOOptions(), &num_hard_links, nullptr)
+      .PermitUncheckedError();
+
+  // We can tolerate rare races where we might immediately delete both links
+  // to a file.
+  if (rate_bytes_per_sec_.load() <= 0 || (!force_bg && num_hard_links > 1)) {
+    Status s = DeleteFileImmediately(file_path, /*accounted=*/false);
+    if (s.ok()) {
+      ROCKS_LOG_INFO(info_log_,
+                     "Deleted file %s immediately, rate_bytes_per_sec %" PRIi64,
+                     file_path.c_str(), rate_bytes_per_sec_.load());
+    }
+    return s;
+  }
+  return AddFileToDeletionQueue(file_path, dir_to_sync, bucket,
+                                /*accounted=*/false);
+}
+
+Status DeleteScheduler::DeleteFileImmediately(const std::string& file_path,
+                                              bool accounted) {
+  TEST_SYNC_POINT("DeleteScheduler::DeleteFile");
+  TEST_SYNC_POINT_CALLBACK("DeleteScheduler::DeleteFile::cb",
+                           const_cast<std::string*>(&file_path));
+  Status s = fs_->DeleteFile(file_path, IOOptions(), nullptr);
+  if (s.ok()) {
+    s = OnDeleteFile(file_path, accounted);
+    InstrumentedMutexLock l(&mu_);
+    RecordTick(stats_.get(), FILES_DELETED_IMMEDIATELY);
+  }
+  return s;
+}
+
+Status DeleteScheduler::AddFileToDeletionQueue(const std::string& file_path,
+                                               const std::string& dir_to_sync,
+                                               std::optional<int32_t> bucket,
+                                               bool accounted) {
   // Move file to trash
   std::string trash_file;
-  Status s = MarkAsTrash(file_path, &trash_file);
+  Status s = MarkAsTrash(file_path, accounted, &trash_file);
   ROCKS_LOG_INFO(info_log_, "Mark file: %s as trash -- %s", trash_file.c_str(),
                  s.ToString().c_str());
 
@@ -94,7 +135,7 @@ Status DeleteScheduler::DeleteFile(const std::string& file_path,
                     file_path.c_str(), s.ToString().c_str());
     s = fs_->DeleteFile(file_path, IOOptions(), nullptr);
     if (s.ok()) {
-      s = sst_file_manager_->OnDeleteFile(file_path);
+      s = OnDeleteFile(file_path, accounted);
       ROCKS_LOG_INFO(info_log_, "Deleted file %s immediately",
                      trash_file.c_str());
       InstrumentedMutexLock l(&mu_);
@@ -104,11 +145,13 @@ Status DeleteScheduler::DeleteFile(const std::string& file_path,
   }
 
   // Update the total trash size
-  uint64_t trash_file_size = 0;
-  IOStatus io_s =
-      fs_->GetFileSize(trash_file, IOOptions(), &trash_file_size, nullptr);
-  if (io_s.ok()) {
-    total_trash_size_.fetch_add(trash_file_size);
+  if (accounted) {
+    uint64_t trash_file_size = 0;
+    IOStatus io_s =
+        fs_->GetFileSize(trash_file, IOOptions(), &trash_file_size, nullptr);
+    if (io_s.ok()) {
+      total_trash_size_.fetch_add(trash_file_size);
+    }
   }
   //**TODO: What should we do if we failed to
   // get the file size?
@@ -117,8 +160,15 @@ Status DeleteScheduler::DeleteFile(const std::string& file_path,
   {
     InstrumentedMutexLock l(&mu_);
     RecordTick(stats_.get(), FILES_MARKED_TRASH);
-    queue_.emplace(trash_file, dir_to_sync);
+    queue_.emplace(trash_file, dir_to_sync, accounted, bucket);
     pending_files_++;
+    if (bucket.has_value()) {
+      auto iter = pending_files_in_buckets_.find(bucket.value());
+      assert(iter != pending_files_in_buckets_.end());
+      if (iter != pending_files_in_buckets_.end()) {
+        iter->second++;
+      }
+    }
     if (pending_files_ == 1) {
       cv_.SignalAll();
     }
@@ -177,7 +227,7 @@ Status DeleteScheduler::CleanupDirectory(Env* env, SstFileManagerImpl* sfm,
 }
 
 Status DeleteScheduler::MarkAsTrash(const std::string& file_path,
-                                    std::string* trash_file) {
+                                    bool accounted, std::string* trash_file) {
   // Sanity check of the path
   size_t idx = file_path.rfind('/');
   if (idx == std::string::npos || idx == file_path.size() - 1) {
@@ -211,7 +261,7 @@ Status DeleteScheduler::MarkAsTrash(const std::string& file_path,
     }
     cnt++;
   }
-  if (s.ok()) {
+  if (s.ok() && accounted) {
     s = sst_file_manager_->OnMoveFile(file_path, *trash_file);
   }
   return s;
@@ -235,6 +285,8 @@ void DeleteScheduler::BackgroundEmptyTrash() {
     uint64_t total_deleted_bytes = 0;
     int64_t current_delete_rate = rate_bytes_per_sec_.load();
     while (!queue_.empty() && !closing_) {
+      // Satisfy static analysis.
+      std::optional<int32_t> bucket = std::nullopt;
       if (current_delete_rate != rate_bytes_per_sec_.load()) {
         // User changed the delete rate
         current_delete_rate = rate_bytes_per_sec_.load();
@@ -247,14 +299,17 @@ void DeleteScheduler::BackgroundEmptyTrash() {
       // Get new file to delete
       const FileAndDir& fad = queue_.front();
       std::string path_in_trash = fad.fname;
+      std::string dir_to_sync = fad.dir;
+      bool accounted = fad.accounted;
+      bucket = fad.bucket;
 
       // We don't need to hold the lock while deleting the file
       mu_.Unlock();
       uint64_t deleted_bytes = 0;
       bool is_complete = true;
       // Delete file from trash and update total_penlty value
-      Status s =
-          DeleteTrashFile(path_in_trash, fad.dir, &deleted_bytes, &is_complete);
+      Status s = DeleteTrashFile(path_in_trash, dir_to_sync, accounted,
+                                 &deleted_bytes, &is_complete);
       total_deleted_bytes += deleted_bytes;
       mu_.Lock();
       if (is_complete) {
@@ -288,12 +343,20 @@ void DeleteScheduler::BackgroundEmptyTrash() {
       TEST_SYNC_POINT_CALLBACK("DeleteScheduler::BackgroundEmptyTrash:Wait",
                                &total_penalty);
 
+      int32_t pending_files_in_bucket = std::numeric_limits<int32_t>::max();
       if (is_complete) {
         pending_files_--;
+        if (bucket.has_value()) {
+          auto iter = pending_files_in_buckets_.find(bucket.value());
+          assert(iter != pending_files_in_buckets_.end());
+          if (iter != pending_files_in_buckets_.end()) {
+            pending_files_in_bucket = iter->second--;
+          }
+        }
       }
-      if (pending_files_ == 0) {
-        // Unblock WaitForEmptyTrash since there are no more files waiting
-        // to be deleted
+      if (pending_files_ == 0 || pending_files_in_bucket == 0) {
+        // Unblock WaitForEmptyTrash or WaitForEmptyTrashBucket since there are
+        // no more files waiting to be deleted
         cv_.SignalAll();
       }
     }
@@ -302,12 +365,14 @@ void DeleteScheduler::BackgroundEmptyTrash() {
 
 Status DeleteScheduler::DeleteTrashFile(const std::string& path_in_trash,
                                         const std::string& dir_to_sync,
-                                        uint64_t* deleted_bytes,
+                                        bool accounted, uint64_t* deleted_bytes,
                                         bool* is_complete) {
   uint64_t file_size;
   Status s = fs_->GetFileSize(path_in_trash, IOOptions(), &file_size, nullptr);
   *is_complete = true;
   TEST_SYNC_POINT("DeleteScheduler::DeleteTrashFile:DeleteFile");
+  TEST_SYNC_POINT_CALLBACK("DeleteScheduler::DeleteTrashFile::cb",
+                           const_cast<std::string*>(&path_in_trash));
   if (s.ok()) {
     bool need_full_delete = true;
     if (bytes_max_delete_chunk_ != 0 && file_size > bytes_max_delete_chunk_) {
@@ -374,7 +439,7 @@ Status DeleteScheduler::DeleteTrashFile(const std::string& path_in_trash,
       }
       if (s.ok()) {
         *deleted_bytes = file_size;
-        s = sst_file_manager_->OnDeleteFile(path_in_trash);
+        s = OnDeleteFile(path_in_trash, accounted);
       }
     }
   }
@@ -384,12 +449,24 @@ Status DeleteScheduler::DeleteTrashFile(const std::string& path_in_trash,
                     path_in_trash.c_str(), s.ToString().c_str());
     *deleted_bytes = 0;
   } else {
-    total_trash_size_.fetch_sub(*deleted_bytes);
+    if (accounted) {
+      total_trash_size_.fetch_sub(*deleted_bytes);
+    }
   }
 
   return s;
 }
 
+Status DeleteScheduler::OnDeleteFile(const std::string& file_path,
+                                     bool accounted) {
+  if (accounted) {
+    return sst_file_manager_->OnDeleteFile(file_path);
+  }
+  TEST_SYNC_POINT_CALLBACK("DeleteScheduler::OnDeleteFile",
+                           const_cast<std::string*>(&file_path));
+  return Status::OK();
+}
+
 void DeleteScheduler::WaitForEmptyTrash() {
   InstrumentedMutexLock l(&mu_);
   while (pending_files_ > 0 && !closing_) {
@@ -397,6 +474,30 @@ void DeleteScheduler::WaitForEmptyTrash() {
   }
 }
 
+std::optional<int32_t> DeleteScheduler::NewTrashBucket() {
+  if (rate_bytes_per_sec_.load() <= 0) {
+    return std::nullopt;
+  }
+  InstrumentedMutexLock l(&mu_);
+  int32_t bucket_number = next_trash_bucket_++;
+  pending_files_in_buckets_.emplace(bucket_number, 0);
+  return bucket_number;
+}
+
+void DeleteScheduler::WaitForEmptyTrashBucket(int32_t bucket) {
+  InstrumentedMutexLock l(&mu_);
+  if (bucket >= next_trash_bucket_) {
+    return;
+  }
+  auto iter = pending_files_in_buckets_.find(bucket);
+  while (iter != pending_files_in_buckets_.end() && iter->second > 0 &&
+         !closing_) {
+    cv_.Wait();
+    iter = pending_files_in_buckets_.find(bucket);
+  }
+  pending_files_in_buckets_.erase(bucket);
+}
+
 void DeleteScheduler::MaybeCreateBackgroundThread() {
   if (bg_thread_ == nullptr && rate_bytes_per_sec_.load() > 0) {
     bg_thread_.reset(

file/delete_scheduler.h

@@ -7,6 +7,7 @@
 
 
 #include <map>
+#include <optional>
 #include <queue>
 #include <string>
 #include <thread>
@@ -48,16 +49,45 @@ class DeleteScheduler {
     MaybeCreateBackgroundThread();
   }
 
-  // Mark file as trash directory and schedule its deletion. If force_bg is
-  // set, it forces the file to always be deleted in the background thread,
-  // except when rate limiting is disabled
+  // Delete an accounted file that is tracked by `SstFileManager` and should be
+  // tracked by this `DeleteScheduler` when it's deleted.
+  // The file is deleted immediately if slow deletion is disabled. If force_bg
+  // is not set and trash to db size ratio exceeded the configured threshold,
+  // it is immediately deleted too. In all other cases, the file will be moved
+  // to a trash directory and scheduled for deletion by a background thread.
   Status DeleteFile(const std::string& fname, const std::string& dir_to_sync,
                     const bool force_bg = false);
 
-  // Wait for all files being deleteing in the background to finish or for
+  // Delete an unaccounted file that is not tracked by `SstFileManager` and
+  // should not be tracked by this `DeleteScheduler` when it's deleted.
+  // The file is deleted immediately if slow deletion is disabled. If force_bg
+  // is not set and the file have more than 1 hard link, it is immediately
+  // deleted too. In all other cases, the file will be moved to a trash
+  // directory and scheduled for deletion by a background thread.
+  // This API also supports assign a file to a specified bucket created by
+  // `NewTrashBucket` when delete files in the background. So the caller can
+  // wait for a specific bucket to be empty by checking the
+  // `WaitForEmptyTrashBucket` API.
+  Status DeleteUnaccountedFile(const std::string& file_path,
+                               const std::string& dir_to_sync,
+                               const bool force_bg = false,
+                               std::optional<int32_t> bucket = std::nullopt);
+
+  // Wait for all files being deleted in the background to finish or for
   // destructor to be called.
   void WaitForEmptyTrash();
 
+  // Creates a new trash bucket. A bucket is only created and returned when slow
+  // deletion is enabled.
+  // For each bucket that is created, the user should also call
+  // `WaitForEmptyTrashBucket` after scheduling file deletions to make sure the
+  // trash files are all cleared.
+  std::optional<int32_t> NewTrashBucket();
+
+  // Wait for all the files in the specified bucket to be deleted in the
+  // background or for the destructor to be called.
+  void WaitForEmptyTrashBucket(int32_t bucket);
+
   // Return a map containing errors that happened in BackgroundEmptyTrash
   // file_path => error status
   std::map<std::string, Status> GetBackgroundErrors();
@@ -87,12 +117,21 @@ class DeleteScheduler {
   }
 
  private:
-  Status MarkAsTrash(const std::string& file_path, std::string* path_in_trash);
+  Status DeleteFileImmediately(const std::string& file_path, bool accounted);
+
+  Status AddFileToDeletionQueue(const std::string& file_path,
+                                const std::string& dir_to_sync,
+                                std::optional<int32_t> bucket, bool accounted);
+
+  Status MarkAsTrash(const std::string& file_path, bool accounted,
+                     std::string* path_in_trash);
 
   Status DeleteTrashFile(const std::string& path_in_trash,
-                         const std::string& dir_to_sync,
+                         const std::string& dir_to_sync, bool accounted,
                          uint64_t* deleted_bytes, bool* is_complete);
 
+  Status OnDeleteFile(const std::string& file_path, bool accounted);
+
   void BackgroundEmptyTrash();
 
   void MaybeCreateBackgroundThread();
@@ -104,19 +143,28 @@ class DeleteScheduler {
   std::atomic<uint64_t> total_trash_size_;
   // Maximum number of bytes that should be deleted per second
   std::atomic<int64_t> rate_bytes_per_sec_;
-  // Mutex to protect queue_, pending_files_, bg_errors_, closing_, stats_
+  // Mutex to protect queue_, pending_files_, next_trash_bucket_,
+  // pending_files_in_buckets_, bg_errors_, closing_, stats_
   InstrumentedMutex mu_;
 
   struct FileAndDir {
-    FileAndDir(const std::string& f, const std::string& d) : fname(f), dir(d) {}
+    FileAndDir(const std::string& _fname, const std::string& _dir,
+               bool _accounted, std::optional<int32_t> _bucket)
+        : fname(_fname), dir(_dir), accounted(_accounted), bucket(_bucket) {}
     std::string fname;
     std::string dir;  // empty will be skipped.
+    bool accounted;
+    std::optional<int32_t> bucket;
   };
 
   // Queue of trash files that need to be deleted
   std::queue<FileAndDir> queue_;
   // Number of trash files that are waiting to be deleted
   int32_t pending_files_;
+  // Next trash bucket that can be created
+  int32_t next_trash_bucket_;
+  // A mapping from trash bucket to number of pending files in the bucket
+  std::map<int32_t, int32_t> pending_files_in_buckets_;
   uint64_t bytes_max_delete_chunk_;
   // Errors that happened in BackgroundEmptyTrash (file_path => error)
   std::map<std::string, Status> bg_errors_;
@@ -127,6 +175,7 @@ class DeleteScheduler {
   // Condition variable signaled in these conditions
   //    - pending_files_ value change from 0 => 1
   //    - pending_files_ value change from 1 => 0
+  //    - a value in pending_files_in_buckets change from 1 => 0
   //    - closing_ value is set to true
   InstrumentedCondVar cv_;
   // Background thread running BackgroundEmptyTrash
@@ -138,6 +187,10 @@ class DeleteScheduler {
   // If the trash size constitutes for more than this fraction of the total DB
   // size we will start deleting new files passed to DeleteScheduler
   // immediately
+  // Unaccounted files passed for deletion will not cause change in
+  // total_trash_size_ or affect the DeleteScheduler::total_trash_size_ over
+  // SstFileManager::total_size_ ratio. Their slow deletion is not subject to
+  // this configured threshold either.
   std::atomic<double> max_trash_db_ratio_;
   static const uint64_t kMicrosInSecond = 1000 * 1000LL;
   std::shared_ptr<Statistics> stats_;

file/delete_scheduler_test.cc

@@ -78,7 +78,7 @@ class DeleteSchedulerTest : public testing::Test {
   }
 
   std::string NewDummyFile(const std::string& file_name, uint64_t size = 1024,
-                           size_t dummy_files_dirs_idx = 0) {
+                           size_t dummy_files_dirs_idx = 0, bool track = true) {
     std::string file_path =
         dummy_files_dirs_[dummy_files_dirs_idx] + "/" + file_name;
     std::unique_ptr<WritableFile> f;
@@ -86,7 +86,9 @@ class DeleteSchedulerTest : public testing::Test {
     std::string data(size, 'A');
     EXPECT_OK(f->Append(data));
     EXPECT_OK(f->Close());
-    EXPECT_OK(sst_file_mgr_->OnAddFile(file_path));
+    if (track) {
+      EXPECT_OK(sst_file_mgr_->OnAddFile(file_path));
+    }
     return file_path;
   }
 
@@ -353,6 +355,8 @@ TEST_F(DeleteSchedulerTest, DisableRateLimiting) {
   ASSERT_EQ(num_files,
             stats_->getAndResetTickerCount(FILES_DELETED_IMMEDIATELY));
 
+  ASSERT_FALSE(delete_scheduler_->NewTrashBucket().has_value());
+
   ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->DisableProcessing();
 }
 
@@ -718,6 +722,141 @@ TEST_F(DeleteSchedulerTest, IsTrashCheck) {
   ASSERT_FALSE(DeleteScheduler::IsTrashFile("abc.trashx"));
 }
 
+TEST_F(DeleteSchedulerTest, DeleteAccountedAndUnaccountedFiles) {
+  rate_bytes_per_sec_ = 1024 * 1024;  // 1 MB / s
+  NewDeleteScheduler();
+
+  // Create 100 files, every file is 1 KB
+  int num_files = 100;        // 100 files
+  uint64_t file_size = 1024;  // 1 KB as a file size
+  std::vector<std::string> generated_files;
+  for (int i = 0; i < num_files; i++) {
+    std::string file_name = "file" + std::to_string(i) + ".data";
+    generated_files.push_back(NewDummyFile(file_name, file_size,
+                                           /*dummy_files_dirs_idx*/ 0,
+                                           /*track=*/false));
+  }
+
+  for (int i = 0; i < num_files; i++) {
+    if (i % 2) {
+      ASSERT_OK(sst_file_mgr_->OnAddFile(generated_files[i], file_size));
+      ASSERT_OK(delete_scheduler_->DeleteFile(generated_files[i], ""));
+    } else {
+      ASSERT_OK(
+          delete_scheduler_->DeleteUnaccountedFile(generated_files[i], ""));
+    }
+  }
+
+  delete_scheduler_->WaitForEmptyTrash();
+  ASSERT_EQ(0, delete_scheduler_->GetTotalTrashSize());
+  ASSERT_EQ(0, sst_file_mgr_->GetTotalSize());
+}
+
+TEST_F(DeleteSchedulerTest, ConcurrentlyDeleteUnaccountedFilesInBuckets) {
+  int bg_delete_file = 0;
+  int fg_delete_file = 0;
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack(
+      "DeleteScheduler::DeleteTrashFile:DeleteFile",
+      [&](void* /*arg*/) { bg_delete_file++; });
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack(
+      "DeleteScheduler::DeleteFile", [&](void* /*arg*/) { fg_delete_file++; });
+  rate_bytes_per_sec_ = 1024 * 1024;  // 1 MB / s
+  NewDeleteScheduler();
+
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->EnableProcessing();
+  // Create 1000 files, every file is 1 KB
+  int num_files = 1000;
+  uint64_t file_size = 1024;  // 1 KB as a file size
+  std::vector<std::string> generated_files;
+  for (int i = 0; i < num_files; i++) {
+    std::string file_name = "file" + std::to_string(i) + ".data";
+    generated_files.push_back(NewDummyFile(file_name, file_size,
+                                           /*dummy_files_dirs_idx*/ 0,
+                                           /*track=*/false));
+  }
+  // Concurrently delete files in different buckets and check all the buckets
+  // are empty.
+  int thread_cnt = 10;
+  int files_per_thread = 100;
+  std::atomic<int> thread_num(0);
+  std::vector<port::Thread> threads;
+  std::function<void()> delete_thread = [&]() {
+    std::optional<int32_t> bucket = delete_scheduler_->NewTrashBucket();
+    ASSERT_TRUE(bucket.has_value());
+    int idx = thread_num.fetch_add(1);
+    int range_start = idx * files_per_thread;
+    int range_end = range_start + files_per_thread;
+    for (int j = range_start; j < range_end; j++) {
+      ASSERT_OK(delete_scheduler_->DeleteUnaccountedFile(
+          generated_files[j], "", /*false_bg=*/false, bucket));
+    }
+    delete_scheduler_->WaitForEmptyTrashBucket(bucket.value());
+  };
+
+  for (int i = 0; i < thread_cnt; i++) {
+    threads.emplace_back(delete_thread);
+  }
+
+  for (size_t i = 0; i < threads.size(); i++) {
+    threads[i].join();
+  }
+
+  ASSERT_EQ(0, delete_scheduler_->GetTotalTrashSize());
+  ASSERT_EQ(0, stats_->getAndResetTickerCount(FILES_DELETED_IMMEDIATELY));
+  ASSERT_EQ(1000, stats_->getAndResetTickerCount(FILES_MARKED_TRASH));
+  ASSERT_EQ(0, fg_delete_file);
+  ASSERT_EQ(1000, bg_delete_file);
+
+  // OK to re check an already empty bucket
+  delete_scheduler_->WaitForEmptyTrashBucket(9);
+  // Invalid bucket return too.
+  delete_scheduler_->WaitForEmptyTrashBucket(100);
+  std::optional<int32_t> next_bucket = delete_scheduler_->NewTrashBucket();
+  ASSERT_TRUE(next_bucket.has_value());
+  ASSERT_EQ(10, next_bucket.value());
+  delete_scheduler_->WaitForEmptyTrashBucket(10);
+
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->DisableProcessing();
+}
+
+TEST_F(DeleteSchedulerTest,
+       ImmediatelyDeleteUnaccountedFilesWithRemainingLinks) {
+  int bg_delete_file = 0;
+  int fg_delete_file = 0;
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack(
+      "DeleteScheduler::DeleteTrashFile:DeleteFile",
+      [&](void* /*arg*/) { bg_delete_file++; });
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack(
+      "DeleteScheduler::DeleteFile", [&](void* /*arg*/) { fg_delete_file++; });
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->EnableProcessing();
+
+  rate_bytes_per_sec_ = 1024 * 1024;  // 1 MB / sec
+  NewDeleteScheduler();
+
+  std::string file1 = NewDummyFile("data_1", 500 * 1024,
+                                   /*dummy_files_dirs_idx*/ 0, /*track=*/false);
+  std::string file2 = NewDummyFile("data_2", 100 * 1024,
+                                   /*dummy_files_dirs_idx*/ 0, /*track=*/false);
+
+  ASSERT_OK(env_->LinkFile(file1, dummy_files_dirs_[0] + "/data_1b"));
+  ASSERT_OK(env_->LinkFile(file2, dummy_files_dirs_[0] + "/data_2b"));
+
+  // Should delete in 4 batch if there is no hardlink
+  ASSERT_OK(
+      delete_scheduler_->DeleteUnaccountedFile(file1, "", /*force_bg=*/false));
+  ASSERT_OK(
+      delete_scheduler_->DeleteUnaccountedFile(file2, "", /*force_bg=*/false));
+
+  delete_scheduler_->WaitForEmptyTrash();
+
+  ASSERT_EQ(0, delete_scheduler_->GetTotalTrashSize());
+  ASSERT_EQ(0, bg_delete_file);
+  ASSERT_EQ(2, fg_delete_file);
+  ASSERT_EQ(0, stats_->getAndResetTickerCount(FILES_MARKED_TRASH));
+  ASSERT_EQ(2, stats_->getAndResetTickerCount(FILES_DELETED_IMMEDIATELY));
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->EnableProcessing();
+}
+
 }  // namespace ROCKSDB_NAMESPACE
 
 int main(int argc, char** argv) {

file/file_util.cc

@@ -125,8 +125,8 @@ IOStatus CreateFile(FileSystem* fs, const std::string& destination,
 Status DeleteDBFile(const ImmutableDBOptions* db_options,
                     const std::string& fname, const std::string& dir_to_sync,
                     const bool force_bg, const bool force_fg) {
-  SstFileManagerImpl* sfm =
-      static_cast<SstFileManagerImpl*>(db_options->sst_file_manager.get());
+  SstFileManagerImpl* sfm = static_cast_with_check<SstFileManagerImpl>(
+      db_options->sst_file_manager.get());
   if (sfm && !force_fg) {
     return sfm->ScheduleFileDeletion(fname, dir_to_sync, force_bg);
   } else {
@@ -134,6 +134,21 @@ Status DeleteDBFile(const ImmutableDBOptions* db_options,
   }
 }
 
+Status DeleteUnaccountedDBFile(const ImmutableDBOptions* db_options,
+                               const std::string& fname,
+                               const std::string& dir_to_sync,
+                               const bool force_bg, const bool force_fg,
+                               std::optional<int32_t> bucket) {
+  SstFileManagerImpl* sfm = static_cast_with_check<SstFileManagerImpl>(
+      db_options->sst_file_manager.get());
+  if (sfm && !force_fg) {
+    return sfm->ScheduleUnaccountedFileDeletion(fname, dir_to_sync, force_bg,
+                                                bucket);
+  } else {
+    return db_options->env->DeleteFile(fname);
+  }
+}
+
 // requested_checksum_func_name brings the function name of the checksum
 // generator in checksum_factory. Empty string is permitted, in which case the
 // name of the generator created by the factory is unchecked. When

file/file_util.h

@@ -55,6 +55,16 @@ Status DeleteDBFile(const ImmutableDBOptions* db_options,
                     const std::string& fname, const std::string& path_to_sync,
                     const bool force_bg, const bool force_fg);
 
+// Delete an unaccounted DB file that is not tracked by SstFileManager and will
+// not be tracked by its DeleteScheduler when getting deleted.
+// If a legitimate bucket is provided and this file is scheduled for slow
+// deletion, it will be assigned to the specified trash bucket.
+Status DeleteUnaccountedDBFile(const ImmutableDBOptions* db_options,
+                               const std::string& fname,
+                               const std::string& dir_to_sync,
+                               const bool force_bg, const bool force_fg,
+                               std::optional<int32_t> bucket);
+
 // TODO(hx235): pass the whole DBOptions intead of its individual fields
 IOStatus GenerateOneFileChecksum(
     FileSystem* fs, const std::string& file_path,

file/sst_file_manager_impl.cc

@@ -421,10 +421,28 @@ Status SstFileManagerImpl::ScheduleFileDeletion(const std::string& file_path,
   return delete_scheduler_.DeleteFile(file_path, path_to_sync, force_bg);
 }
 
+Status SstFileManagerImpl::ScheduleUnaccountedFileDeletion(
+    const std::string& file_path, const std::string& dir_to_sync,
+    const bool force_bg, std::optional<int32_t> bucket) {
+  TEST_SYNC_POINT_CALLBACK(
+      "SstFileManagerImpl::ScheduleUnaccountedFileDeletion",
+      const_cast<std::string*>(&file_path));
+  return delete_scheduler_.DeleteUnaccountedFile(file_path, dir_to_sync,
+                                                 force_bg, bucket);
+}
+
 void SstFileManagerImpl::WaitForEmptyTrash() {
   delete_scheduler_.WaitForEmptyTrash();
 }
 
+std::optional<int32_t> SstFileManagerImpl::NewTrashBucket() {
+  return delete_scheduler_.NewTrashBucket();
+}
+
+void SstFileManagerImpl::WaitForEmptyTrashBucket(int32_t bucket) {
+  delete_scheduler_.WaitForEmptyTrashBucket(bucket);
+}
+
 void SstFileManagerImpl::OnAddFileImpl(const std::string& file_path,
                                        uint64_t file_size) {
   auto tracked_file = tracked_files_.find(file_path);

file/sst_file_manager_impl.h

@@ -5,7 +5,7 @@
 
 #pragma once
 
-
+#include <optional>
 #include <string>
 
 #include "db/compaction/compaction.h"
@@ -118,17 +118,40 @@ class SstFileManagerImpl : public SstFileManager {
   // not guaranteed
   bool CancelErrorRecovery(ErrorHandler* db);
 
-  // Mark file as trash and schedule it's deletion. If force_bg is set, it
+  // Mark a file as trash and schedule its deletion. If force_bg is set, it
   // forces the file to be deleting in the background regardless of DB size,
-  // except when rate limited delete is disabled
+  // except when rate limited delete is disabled.
   virtual Status ScheduleFileDeletion(const std::string& file_path,
                                       const std::string& dir_to_sync,
                                       const bool force_bg = false);
 
-  // Wait for all files being deleteing in the background to finish or for
+  // Delete an unaccounted file. The file is deleted immediately if slow
+  // deletion is disabled. A file with more than 1 hard links will be deleted
+  // immediately unless force_bg is set. In other cases, files will be scheduled
+  // for slow deletion, and assigned to the specified bucket if a legitimate one
+  // is provided. A legitimate bucket is one that is created with the
+  // `NewTrashBucket` API, and for which `WaitForEmptyTrashBucket` hasn't been
+  // called yet.
+  virtual Status ScheduleUnaccountedFileDeletion(
+      const std::string& file_path, const std::string& dir_to_sync,
+      const bool force_bg = false,
+      std::optional<int32_t> bucket = std::nullopt);
+
+  // Wait for all files being deleted in the background to finish or for
   // destructor to be called.
   virtual void WaitForEmptyTrash();
 
+  // Creates a new trash bucket. A legitimate bucket is only created and
+  // returned when slow deletion is enabled.
+  // For each bucket that is created and used, the user should also call
+  // `WaitForEmptyTrashBucket` after scheduling file deletions to make sure all
+  // the trash files are cleared.
+  std::optional<int32_t> NewTrashBucket();
+
+  // Wait for all the files in the specified bucket to be deleted in the
+  // background or for destructor to be called.
+  virtual void WaitForEmptyTrashBucket(int32_t bucket);
+
   DeleteScheduler* delete_scheduler() { return &delete_scheduler_; }
 
   // Stop the error recovery background thread. This should be called only

include/rocksdb/statistics.h

@@ -529,6 +529,11 @@ enum Tickers : uint32_t {
   // Footer corruption detected when opening an SST file for reading
   SST_FOOTER_CORRUPTION_COUNT,
 
+  // Counters for file read retries with the verify_and_reconstruct_read
+  // file system option after detecting a checksum mismatch
+  FILE_READ_CORRUPTION_RETRY_COUNT,
+  FILE_READ_CORRUPTION_RETRY_SUCCESS_COUNT,
+
   TICKER_ENUM_MAX
 };
 

include/rocksdb/version.h

@@ -13,7 +13,7 @@
 // minor or major version number planned for release.
 #define ROCKSDB_MAJOR 9
 #define ROCKSDB_MINOR 5
-#define ROCKSDB_PATCH 0
+#define ROCKSDB_PATCH 2
 
 // 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

java/rocksjni/portal.h

@@ -5269,6 +5269,10 @@ class TickerTypeJni {
         return -0x53;
       case ROCKSDB_NAMESPACE::Tickers::SST_FOOTER_CORRUPTION_COUNT:
         return -0x55;
+      case ROCKSDB_NAMESPACE::Tickers::FILE_READ_CORRUPTION_RETRY_COUNT:
+        return -0x56;
+      case ROCKSDB_NAMESPACE::Tickers::FILE_READ_CORRUPTION_RETRY_SUCCESS_COUNT:
+        return -0x57;
       case ROCKSDB_NAMESPACE::Tickers::TICKER_ENUM_MAX:
         // -0x54 is the max value at this time. Since these values are exposed
         // directly to Java clients, we'll keep the value the same till the next
@@ -5726,6 +5730,11 @@ class TickerTypeJni {
         return ROCKSDB_NAMESPACE::Tickers::PREFETCH_HITS;
       case -0x55:
         return ROCKSDB_NAMESPACE::Tickers::SST_FOOTER_CORRUPTION_COUNT;
+      case -0x56:
+        return ROCKSDB_NAMESPACE::Tickers::FILE_READ_CORRUPTION_RETRY_COUNT;
+      case -0x57:
+        return ROCKSDB_NAMESPACE::Tickers::
+            FILE_READ_CORRUPTION_RETRY_SUCCESS_COUNT;
       case -0x54:
         // -0x54 is the max value at this time. Since these values are exposed
         // directly to Java clients, we'll keep the value the same till the next

java/src/main/java/org/rocksdb/TickerType.java

@@ -878,6 +878,10 @@ public enum TickerType {
 
     SST_FOOTER_CORRUPTION_COUNT((byte) -0x55),
 
+    FILE_READ_CORRUPTION_RETRY_COUNT((byte) -0x56),
+
+    FILE_READ_CORRUPTION_RETRY_SUCCESS_COUNT((byte) -0x57),
+
     TICKER_ENUM_MAX((byte) -0x54);
 
     private final byte value;

monitoring/statistics.cc

@@ -266,6 +266,10 @@ const std::vector<std::pair<Tickers, std::string>> TickersNameMap = {
     {PREFETCH_BYTES_USEFUL, "rocksdb.prefetch.bytes.useful"},
     {PREFETCH_HITS, "rocksdb.prefetch.hits"},
     {SST_FOOTER_CORRUPTION_COUNT, "rocksdb.footer.corruption.count"},
+    {FILE_READ_CORRUPTION_RETRY_COUNT,
+     "rocksdb.file.read.corruption.retry.count"},
+    {FILE_READ_CORRUPTION_RETRY_SUCCESS_COUNT,
+     "rocksdb.file.read.corruption.retry.success.count"},
 };
 
 const std::vector<std::pair<Histograms, std::string>> HistogramsNameMap = {

table/block_based/block_based_table_reader.cc

@@ -693,6 +693,10 @@ Status BlockBasedTable::Open(
       s = ReadFooterFromFile(retry_opts, file.get(), *ioptions.fs,
                              prefetch_buffer.get(), file_size, &footer,
                              kBlockBasedTableMagicNumber);
+      RecordTick(ioptions.stats, FILE_READ_CORRUPTION_RETRY_COUNT);
+      if (s.ok()) {
+        RecordTick(ioptions.stats, FILE_READ_CORRUPTION_RETRY_SUCCESS_COUNT);
+      }
     }
   }
   if (!s.ok()) {

table/block_based/block_based_table_reader_sync_and_async.h

@@ -223,13 +223,16 @@ DEFINE_SYNC_AND_ASYNC(void, BlockBasedTable::RetrieveMultipleBlocks)
         s = VerifyBlockChecksum(footer, data, handle.size(),
                                 rep_->file->file_name(), handle.offset());
         RecordTick(ioptions.stats, BLOCK_CHECKSUM_COMPUTE_COUNT);
+        if (!s.ok()) {
+          RecordTick(ioptions.stats, BLOCK_CHECKSUM_MISMATCH_COUNT);
+        }
         TEST_SYNC_POINT_CALLBACK("RetrieveMultipleBlocks:VerifyChecksum", &s);
         if (!s.ok() &&
             CheckFSFeatureSupport(ioptions.fs.get(),
                                   FSSupportedOps::kVerifyAndReconstructRead)) {
           assert(s.IsCorruption());
           assert(!ioptions.allow_mmap_reads);
-          RecordTick(ioptions.stats, BLOCK_CHECKSUM_MISMATCH_COUNT);
+          RecordTick(ioptions.stats, FILE_READ_CORRUPTION_RETRY_COUNT);
 
           // Repeat the read for this particular block using the regular
           // synchronous Read API. We can use the same chunk of memory
@@ -246,6 +249,10 @@ DEFINE_SYNC_AND_ASYNC(void, BlockBasedTable::RetrieveMultipleBlocks)
             assert(result.size() == BlockSizeWithTrailer(handle));
             s = VerifyBlockChecksum(footer, data, handle.size(),
                                     rep_->file->file_name(), handle.offset());
+            if (s.ok()) {
+              RecordTick(ioptions.stats,
+                         FILE_READ_CORRUPTION_RETRY_SUCCESS_COUNT);
+            }
           } else {
             s = io_s;
           }

table/block_fetcher.cc

@@ -84,7 +84,9 @@ inline bool BlockFetcher::TryGetFromPrefetchBuffer() {
         if (io_status_.ok()) {
           got_from_prefetch_buffer_ = true;
           used_buf_ = const_cast<char*>(slice_.data());
-        } else if (!(io_status_.IsCorruption() && retry_corrupt_read_)) {
+        } else if (io_status_.IsCorruption()) {
+          // Returning true apparently indicates we either got some data from
+          // the prefetch buffer, or we tried and encountered an error.
           return true;
         }
       }
@@ -334,9 +336,15 @@ void BlockFetcher::ReadBlock(bool retry) {
     ProcessTrailerIfPresent();
   }
 
+  if (retry) {
+    RecordTick(ioptions_.stats, FILE_READ_CORRUPTION_RETRY_COUNT);
+  }
   if (io_status_.ok()) {
     InsertCompressedBlockToPersistentCacheIfNeeded();
     fs_buf_ = std::move(read_req.fs_scratch);
+    if (retry) {
+      RecordTick(ioptions_.stats, FILE_READ_CORRUPTION_RETRY_SUCCESS_COUNT);
+    }
   } else {
     ReleaseFileSystemProvidedBuffer(&read_req);
     direct_io_buf_.reset();
@@ -355,7 +363,11 @@ IOStatus BlockFetcher::ReadBlockContents() {
     return IOStatus::OK();
   }
   if (TryGetFromPrefetchBuffer()) {
+    if (io_status_.IsCorruption() && retry_corrupt_read_) {
+      ReadBlock(/*retry=*/true);
+    }
     if (!io_status_.ok()) {
+      assert(!fs_buf_);
       return io_status_;
     }
   } else if (!TryGetSerializedBlockFromPersistentCache()) {

unreleased_history/behavior_changes/checkpoint_log_flush_improve.md

@@ -1 +0,0 @@
-When calculating total log size for the `log_size_for_flush` argument in `CreateCheckpoint` API, the size of the archived log will not be included to avoid unnecessary flush

unreleased_history/bug_fixes/0seek_for_prev_prefix.md

@@ -1 +0,0 @@
-* Fix a major bug in which an iterator using prefix filtering and SeekForPrev might miss data when the DB is using `whole_key_filtering=false` and `partition_filters=true`.

unreleased_history/bug_fixes/event_listener_not_called.md

@@ -1 +0,0 @@
-Fixed a bug where `OnErrorRecoveryBegin()` is not called before auto recovery starts.

unreleased_history/bug_fixes/event_listener_race_with_error_handler.md

@@ -1 +0,0 @@
-Fixed a bug where event listener reads ErrorHandler's `bg_error_` member without holding db mutex(#12803).

unreleased_history/bug_fixes/handle_manifest_error.md

@@ -1 +0,0 @@
-Fixed a bug in handling MANIFEST write error that caused the latest valid MANIFEST file to get deleted, resulting in the DB being unopenable.

unreleased_history/bug_fixes/manifest_recovery_ingest_race.md

@@ -1 +0,0 @@
-Fixed a race between error recovery due to manifest sync or write failure and external SST file ingestion. Both attempt to write a new manifest file, which causes an assertion failure.

unreleased_history/performance_improvements/compaction-io-mutex.md

@@ -1 +0,0 @@
-* Fix an issue where compactions were opening table files and reading table properties while holding db mutex_.

unreleased_history/performance_improvements/sorted_wals.md

@@ -1 +0,0 @@
-* Reduce unnecessary filesystem queries and DB mutex acquires in creating backups and checkpoints.

unreleased_history/public_api_changes/expose_writebatch_iterator_cf_methods_via_c_api.md

@@ -1 +0,0 @@
-Introduced new C API function rocksdb_writebatch_iterate_cf for column family-aware iteration over the contents of a WriteBatch

unreleased_history/public_api_changes/ingest_live_db_file.md

@@ -1 +0,0 @@
-* Add support to ingest SST files generated by a DB instead of SstFileWriter. This can be enabled with experimental option `IngestExternalFileOptions::allow_db_generated_files`.

utilities/blob_db/blob_db_test.cc

@@ -760,7 +760,7 @@ TEST_F(BlobDBTest, SstFileManager) {
   // run the same test for Get(), MultiGet() and Iterator each.
   std::shared_ptr<SstFileManager> sst_file_manager(
       NewSstFileManager(mock_env_.get()));
-  sst_file_manager->SetDeleteRateBytesPerSecond(1);
+  sst_file_manager->SetDeleteRateBytesPerSecond(1024 * 1024);
   SstFileManagerImpl *sfm =
       static_cast<SstFileManagerImpl *>(sst_file_manager.get());
 
@@ -818,7 +818,7 @@ TEST_F(BlobDBTest, SstFileManagerRestart) {
   // run the same test for Get(), MultiGet() and Iterator each.
   std::shared_ptr<SstFileManager> sst_file_manager(
       NewSstFileManager(mock_env_.get()));
-  sst_file_manager->SetDeleteRateBytesPerSecond(1);
+  sst_file_manager->SetDeleteRateBytesPerSecond(1024 * 1024);
   SstFileManagerImpl *sfm =
       static_cast<SstFileManagerImpl *>(sst_file_manager.get());
 

utilities/checkpoint/checkpoint_test.cc

@@ -25,6 +25,7 @@
 #include "rocksdb/db.h"
 #include "rocksdb/env.h"
 #include "rocksdb/rocksdb_namespace.h"
+#include "rocksdb/sst_file_manager.h"
 #include "rocksdb/utilities/transaction_db.h"
 #include "test_util/sync_point.h"
 #include "test_util/testharness.h"
@@ -1031,6 +1032,100 @@ TEST_F(CheckpointTest, CheckpointWithArchievedLog) {
   delete snapshot_db;
 }
 
+class CheckpointDestroyTest : public CheckpointTest,
+                              public testing::WithParamInterface<bool> {};
+
+TEST_P(CheckpointDestroyTest, DisableEnableSlowDeletion) {
+  bool slow_deletion = GetParam();
+  Options options = CurrentOptions();
+  options.num_levels = 2;
+  options.disable_auto_compactions = true;
+  Status s;
+  options.sst_file_manager.reset(NewSstFileManager(
+      options.env, options.info_log, "", slow_deletion ? 1024 * 1024 : 0,
+      false /* delete_existing_trash */, &s, 1));
+  ASSERT_OK(s);
+  DestroyAndReopen(options);
+
+  ASSERT_OK(Put("foo", "a"));
+  ASSERT_OK(Flush());
+  ASSERT_OK(Put("bar", "b"));
+  ASSERT_OK(Flush());
+  ASSERT_OK(db_->CompactRange(CompactRangeOptions(), nullptr, nullptr));
+  for (int i = 0; i < 10; i++) {
+    ASSERT_OK(Put("bar", "val" + std::to_string(i)));
+    ASSERT_OK(Flush());
+  }
+  ASSERT_EQ(NumTableFilesAtLevel(0), 10);
+  ASSERT_EQ(NumTableFilesAtLevel(1), 2);
+
+  Checkpoint* checkpoint;
+  ASSERT_OK(Checkpoint::Create(db_, &checkpoint));
+  ASSERT_OK(checkpoint->CreateCheckpoint(snapshot_name_));
+
+  delete checkpoint;
+  checkpoint = nullptr;
+
+  ASSERT_OK(db_->CompactRange(CompactRangeOptions(), nullptr, nullptr));
+  ASSERT_EQ(NumTableFilesAtLevel(0), 0);
+  ASSERT_EQ(NumTableFilesAtLevel(1), 2);
+
+  DB* snapshot_db;
+  ASSERT_OK(DB::Open(options, snapshot_name_, &snapshot_db));
+  ReadOptions read_opts;
+  std::string get_result;
+  ASSERT_OK(snapshot_db->Get(read_opts, "foo", &get_result));
+  ASSERT_EQ("a", get_result);
+  ASSERT_OK(snapshot_db->Get(read_opts, "bar", &get_result));
+  ASSERT_EQ("val9", get_result);
+  delete snapshot_db;
+
+  // Make sure original obsolete files for hard linked files are all deleted.
+  DBImpl* db_impl = static_cast_with_check<DBImpl>(db_);
+  db_impl->TEST_DeleteObsoleteFiles();
+  auto sfm = static_cast_with_check<SstFileManagerImpl>(
+      options.sst_file_manager.get());
+  ASSERT_NE(nullptr, sfm);
+  sfm->WaitForEmptyTrash();
+  // SST file 2-12 for "bar" will be compacted into one file on L1 during the
+  // compaction  after checkpoint is created. SST file 1 on L1: foo, seq:
+  // 1 (hard links is 1 after checkpoint destroy)
+  std::atomic<int> bg_delete_sst{0};
+  std::atomic<int> fg_delete_sst{0};
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack(
+      "DeleteScheduler::DeleteFile::cb", [&](void* arg) {
+        ASSERT_NE(nullptr, arg);
+        auto file_name = *static_cast<std::string*>(arg);
+        if (file_name.size() >= 4 &&
+            file_name.compare(file_name.size() - 4, 4, ".sst") == 0) {
+          fg_delete_sst.fetch_add(1);
+        }
+      });
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack(
+      "DeleteScheduler::DeleteTrashFile::cb", [&](void* arg) {
+        ASSERT_NE(nullptr, arg);
+        auto file_name = *static_cast<std::string*>(arg);
+        if (file_name.size() >= 10 &&
+            file_name.compare(file_name.size() - 10, 10, ".sst.trash") == 0) {
+          bg_delete_sst.fetch_add(1);
+        }
+      });
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->EnableProcessing();
+  ASSERT_OK(DestroyDB(snapshot_name_, options));
+  if (slow_deletion) {
+    ASSERT_EQ(fg_delete_sst, 1);
+    ASSERT_EQ(bg_delete_sst, 11);
+  } else {
+    ASSERT_EQ(fg_delete_sst, 12);
+  }
+
+  ASSERT_EQ("a", Get("foo"));
+  ASSERT_EQ("val9", Get("bar"));
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->DisableProcessing();
+}
+INSTANTIATE_TEST_CASE_P(CheckpointDestroyTest, CheckpointDestroyTest,
+                        ::testing::Values(true, false));
+
 }  // namespace ROCKSDB_NAMESPACE
 
 int main(int argc, char** argv) {

utilities/transactions/pessimistic_transaction.cc

@@ -1227,14 +1227,15 @@ Status PessimisticTransaction::SetName(const TransactionName& name) {
   if (txn_state_ == STARTED) {
     if (name_.length()) {
       s = Status::InvalidArgument("Transaction has already been named.");
-    } else if (txn_db_impl_->GetTransactionByName(name) != nullptr) {
-      s = Status::InvalidArgument("Transaction name must be unique.");
     } else if (name.length() < 1 || name.length() > 512) {
       s = Status::InvalidArgument(
           "Transaction name length must be between 1 and 512 chars.");
     } else {
       name_ = name;
-      txn_db_impl_->RegisterTransaction(this);
+      s = txn_db_impl_->RegisterTransaction(this);
+      if (!s.ok()) {
+        name_.clear();
+      }
     }
   } else {
     s = Status::InvalidArgument("Transaction is beyond state for naming.");

utilities/transactions/pessimistic_transaction_db.cc

@@ -723,6 +723,11 @@ void PessimisticTransactionDB::ReinitializeTransaction(
 Transaction* PessimisticTransactionDB::GetTransactionByName(
     const TransactionName& name) {
   std::lock_guard<std::mutex> lock(name_map_mutex_);
+  return GetTransactionByNameLocked(name);
+}
+
+Transaction* PessimisticTransactionDB::GetTransactionByNameLocked(
+    const TransactionName& name) {
   auto it = transactions_.find(name);
   if (it == transactions_.end()) {
     return nullptr;
@@ -755,13 +760,15 @@ void PessimisticTransactionDB::SetDeadlockInfoBufferSize(uint32_t target_size) {
   lock_manager_->Resize(target_size);
 }
 
-void PessimisticTransactionDB::RegisterTransaction(Transaction* txn) {
+Status PessimisticTransactionDB::RegisterTransaction(Transaction* txn) {
   assert(txn);
   assert(txn->GetName().length() > 0);
-  assert(GetTransactionByName(txn->GetName()) == nullptr);
   assert(txn->GetState() == Transaction::STARTED);
   std::lock_guard<std::mutex> lock(name_map_mutex_);
-  transactions_[txn->GetName()] = txn;
+  if (!transactions_.insert({txn->GetName(), txn}).second) {
+    return Status::InvalidArgument("Duplicate txn name " + txn->GetName());
+  }
+  return Status::OK();
 }
 
 void PessimisticTransactionDB::UnregisterTransaction(Transaction* txn) {

utilities/transactions/pessimistic_transaction_db.h

@@ -173,7 +173,7 @@ class PessimisticTransactionDB : public TransactionDB {
 
   Transaction* GetTransactionByName(const TransactionName& name) override;
 
-  void RegisterTransaction(Transaction* txn);
+  Status RegisterTransaction(Transaction* txn);
   void UnregisterTransaction(Transaction* txn);
 
   // not thread safe. current use case is during recovery (single thread)
@@ -239,6 +239,7 @@ class PessimisticTransactionDB : public TransactionDB {
   friend class WriteUnpreparedTransactionTest_MarkLogWithPrepSection_Test;
 
   Transaction* BeginInternalTransaction(const WriteOptions& options);
+  Transaction* GetTransactionByNameLocked(const TransactionName& name);
 
   std::shared_ptr<LockManager> lock_manager_;