Update HISTORY for 9.3.2

Summary:
The `PessimisticTransaction::SetName()` code checks for an existing txn of the given name before registering the new txn. However, this is not atomic, which could result in a race condition if two txns try to register with the same name. Both might succeed and lead to unpredictable behavior. This PR makes the test and set atomic.

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

Reviewed By: pdillinger

Differential Revision: D60460482

Pulled By: anand1976

fbshipit-source-id: e8afeb2356e1b8f4e8df785cb73532739f82579d

HISTORY.md

@@ -1,6 +1,40 @@
 # Rocksdb Change Log
 > NOTE: Entries for next release do not go here. Follow instructions in `unreleased_history/README.txt`
 
+## 9.3.2 (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`.
+* 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.
+
+## 9.3.1 (05/25/2024)
+### Bug Fixes
+* [internal only] Build script improvement
+
+## 9.3.0 (05/17/2024)
+### New Features
+* Optimistic transactions and pessimistic transactions with the WriteCommitted policy now support the `GetEntity` API.
+* Added new `Iterator` property, "rocksdb.iterator.is-value-pinned", for checking whether the `Slice` returned by `Iterator::value()` can be used until the `Iterator` is destroyed.
+* Optimistic transactions and WriteCommitted pessimistic transactions now support the `MultiGetEntity` API.
+* Optimistic transactions and pessimistic transactions with the WriteCommitted policy now support the `PutEntity` API. Support for read APIs and other write policies (WritePrepared, WriteUnprepared) will be added later.
+
+### Public API Changes
+* Exposed block based metadata cache options via C API
+* Exposed compaction pri via c api.
+* Add a kAdmPolicyAllowAll option to TieredAdmissionPolicy that admits all blocks evicted from the primary block cache into the compressed secondary cache.
+
+### Behavior Changes
+* CompactRange() with change_level=true on a CF with FIFO compaction will return Status::NotSupported().
+* External file ingestion with FIFO compaction will always ingest to L0.
+
+### Bug Fixes
+* Fixed a bug for databases using `DBOptions::allow_2pc == true` (all `TransactionDB`s except `OptimisticTransactionDB`) that have exactly one column family. Due to a missing WAL sync, attempting to open the DB could have returned a `Status::Corruption` with a message like "SST file is ahead of WALs".
+* Fix a bug in CreateColumnFamilyWithImport() where if multiple CFs are imported, we were not resetting files' epoch number and L0 files can have overlapping key range but the same epoch number.
+* Fixed race conditions when `ColumnFamilyOptions::inplace_update_support == true` between user overwrites and reads on the same key.
+* Fix a bug where `CompactFiles()` can compact files of range conflict with other ongoing compactions' when `preclude_last_level_data_seconds > 0` is used
+* Fixed a false positive `Status::Corruption` reported when reopening a DB that used `DBOptions::recycle_log_file_num > 0` and `DBOptions::wal_compression != kNoCompression`.
+* While WAL is locked with LockWAL(), some operations like Flush() and IngestExternalFile() are now blocked as they should have been.
+* Fixed a bug causing stale memory access when using the TieredSecondaryCache with an NVM secondary cache, and a file system that supports return an FS allocated buffer for MultiRead (FSSupportedOps::kFSBuffer is set).
+
 ## 9.2.0 (05/01/2024)
 ### New Features
 * Added two options `deadline` and `max_size_bytes` for CacheDumper to exit early

TARGETS

@@ -4,6 +4,7 @@
 # This file is a Facebook-specific integration for buck builds, so can
 # only be validated by Facebook employees.
 load("//rocks/buckifier:defs.bzl", "cpp_library_wrapper","rocks_cpp_library_wrapper","cpp_binary_wrapper","cpp_unittest_wrapper","fancy_bench_wrapper","add_c_test_wrapper")
+load("@fbcode_macros//build_defs:export_files.bzl", "export_file")
 
 
 cpp_library_wrapper(name="rocksdb_lib", srcs=[

buckifier/targets_cfg.py

@@ -7,6 +7,7 @@ rocksdb_target_header_template = """# This file \100generated by:
 # This file is a Facebook-specific integration for buck builds, so can
 # only be validated by Facebook employees.
 load("//rocks/buckifier:defs.bzl", "cpp_library_wrapper","rocks_cpp_library_wrapper","cpp_binary_wrapper","cpp_unittest_wrapper","fancy_bench_wrapper","add_c_test_wrapper")
+load("@fbcode_macros//build_defs:export_files.bzl", "export_file")
 
 """
 

cache/tiered_secondary_cache_test.cc

@@ -873,6 +873,111 @@ TEST_P(DBTieredAdmPolicyTest, CompressedCacheAdmission) {
   Destroy(options);
 }
 
+TEST_F(DBTieredSecondaryCacheTest, FSBufferTest) {
+  class WrapFS : public FileSystemWrapper {
+   public:
+    explicit WrapFS(const std::shared_ptr<FileSystem>& _target)
+        : FileSystemWrapper(_target) {}
+    ~WrapFS() override {}
+    const char* Name() const override { return "WrapFS"; }
+
+    IOStatus NewRandomAccessFile(const std::string& fname,
+                                 const FileOptions& opts,
+                                 std::unique_ptr<FSRandomAccessFile>* result,
+                                 IODebugContext* dbg) override {
+      class WrappedRandomAccessFile : public FSRandomAccessFileOwnerWrapper {
+       public:
+        explicit WrappedRandomAccessFile(
+            std::unique_ptr<FSRandomAccessFile>& file)
+            : FSRandomAccessFileOwnerWrapper(std::move(file)) {}
+
+        IOStatus MultiRead(FSReadRequest* reqs, size_t num_reqs,
+                           const IOOptions& options,
+                           IODebugContext* dbg) override {
+          for (size_t i = 0; i < num_reqs; ++i) {
+            FSReadRequest& req = reqs[i];
+            FSAllocationPtr buffer(new char[req.len], [](void* ptr) {
+              delete[] static_cast<char*>(ptr);
+            });
+            req.fs_scratch = std::move(buffer);
+            req.status = Read(req.offset, req.len, options, &req.result,
+                              static_cast<char*>(req.fs_scratch.get()), dbg);
+          }
+          return IOStatus::OK();
+        }
+      };
+
+      std::unique_ptr<FSRandomAccessFile> file;
+      IOStatus s = target()->NewRandomAccessFile(fname, opts, &file, dbg);
+      EXPECT_OK(s);
+      result->reset(new WrappedRandomAccessFile(file));
+
+      return s;
+    }
+
+    void SupportedOps(int64_t& supported_ops) override {
+      supported_ops = 1 << FSSupportedOps::kAsyncIO;
+      supported_ops |= 1 << FSSupportedOps::kFSBuffer;
+    }
+  };
+
+  if (!LZ4_Supported()) {
+    ROCKSDB_GTEST_SKIP("This test requires LZ4 support.");
+    return;
+  }
+
+  std::shared_ptr<WrapFS> wrap_fs =
+      std::make_shared<WrapFS>(env_->GetFileSystem());
+  std::unique_ptr<Env> wrap_env(new CompositeEnvWrapper(env_, wrap_fs));
+  BlockBasedTableOptions table_options;
+  table_options.block_cache = NewCache(250 * 1024, 20 * 1024, 256 * 1024,
+                                       TieredAdmissionPolicy::kAdmPolicyAuto,
+                                       /*ready_before_wait=*/true);
+  table_options.block_size = 4 * 1024;
+  table_options.cache_index_and_filter_blocks = false;
+  Options options = GetDefaultOptions();
+  options.create_if_missing = true;
+  options.table_factory.reset(NewBlockBasedTableFactory(table_options));
+  options.statistics = CreateDBStatistics();
+  options.env = wrap_env.get();
+
+  options.paranoid_file_checks = false;
+  DestroyAndReopen(options);
+  Random rnd(301);
+  const int N = 256;
+  for (int i = 0; i < N; i++) {
+    std::string p_v;
+    test::CompressibleString(&rnd, 0.5, 1007, &p_v);
+    ASSERT_OK(Put(Key(i), p_v));
+  }
+
+  ASSERT_OK(Flush());
+
+  std::vector<std::string> keys;
+  std::vector<std::string> values;
+
+  keys.push_back(Key(0));
+  keys.push_back(Key(4));
+  keys.push_back(Key(8));
+  values = MultiGet(keys, /*snapshot=*/nullptr, /*async=*/true);
+  ASSERT_EQ(values.size(), keys.size());
+  for (const auto& value : values) {
+    ASSERT_EQ(1007, value.size());
+  }
+  ASSERT_EQ(nvm_sec_cache()->num_insert_saved(), 3u);
+  ASSERT_EQ(nvm_sec_cache()->num_misses(), 3u);
+  ASSERT_EQ(nvm_sec_cache()->num_hits(), 0u);
+
+  std::string v = Get(Key(12));
+  ASSERT_EQ(1007, v.size());
+  ASSERT_EQ(nvm_sec_cache()->num_insert_saved(), 4u);
+  ASSERT_EQ(nvm_sec_cache()->num_misses(), 4u);
+  ASSERT_EQ(options.statistics->getTickerCount(BLOCK_CACHE_MISS), 4u);
+
+  Close();
+  Destroy(options);
+}
+
 INSTANTIATE_TEST_CASE_P(
     DBTieredAdmPolicyTest, DBTieredAdmPolicyTest,
     ::testing::Values(TieredAdmissionPolicy::kAdmPolicyAuto,

db/db_filesnapshot.cc

@@ -229,12 +229,18 @@ Status DBImpl::GetLiveFilesStorageInfo(
   // metadata.
   mutex_.Lock();
   if (flush_memtable) {
-    Status status = FlushForGetLiveFiles();
-    if (!status.ok()) {
-      mutex_.Unlock();
-      ROCKS_LOG_ERROR(immutable_db_options_.info_log, "Cannot Flush data %s\n",
-                      status.ToString().c_str());
-      return status;
+    bool wal_locked = lock_wal_count_ > 0;
+    if (wal_locked) {
+      ROCKS_LOG_INFO(immutable_db_options_.info_log,
+                     "Can't FlushForGetLiveFiles while WAL is locked");
+    } else {
+      Status status = FlushForGetLiveFiles();
+      if (!status.ok()) {
+        mutex_.Unlock();
+        ROCKS_LOG_ERROR(immutable_db_options_.info_log,
+                        "Cannot Flush data %s\n", status.ToString().c_str());
+        return status;
+      }
     }
   }
 

db/db_impl/db_impl.cc

@@ -17,6 +17,7 @@
 #include <cstdio>
 #include <map>
 #include <memory>
+#include <optional>
 #include <set>
 #include <sstream>
 #include <stdexcept>
@@ -5163,6 +5164,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
@@ -5174,6 +5183,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);
@@ -5190,10 +5200,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);
         }
@@ -5202,6 +5211,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) {
@@ -5223,18 +5233,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;
@@ -5258,46 +5269,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

@@ -1211,6 +1211,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_;
   }
@@ -2054,17 +2056,22 @@ class DBImpl : public DB {
       mutex_.Lock();
     }
 
-    if (!immutable_db_options_.unordered_write) {
-      // Then the writes are finished before the next write group starts
-      return;
+    if (immutable_db_options_.unordered_write) {
+      // Wait for the ones who already wrote to the WAL to finish their
+      // memtable write.
+      if (pending_memtable_writes_.load() != 0) {
+        // XXX: suspicious wait while holding DB mutex?
+        std::unique_lock<std::mutex> guard(switch_mutex_);
+        switch_cv_.wait(guard,
+                        [&] { return pending_memtable_writes_.load() == 0; });
+      }
+    } else {
+      // (Writes are finished before the next write group starts.)
     }
 
-    // Wait for the ones who already wrote to the WAL to finish their
-    // memtable write.
-    if (pending_memtable_writes_.load() != 0) {
-      std::unique_lock<std::mutex> guard(switch_mutex_);
-      switch_cv_.wait(guard,
-                      [&] { return pending_memtable_writes_.load() == 0; });
+    // Wait for any LockWAL to clear
+    while (lock_wal_count_ > 0) {
+      bg_cv_.Wait();
     }
   }
 

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_write.cc

@@ -2156,6 +2156,8 @@ void DBImpl::NotifyOnMemTableSealed(ColumnFamilyData* /*cfd*/,
 // two_write_queues_ is true (This is to simplify the reasoning.)
 Status DBImpl::SwitchMemtable(ColumnFamilyData* cfd, WriteContext* context) {
   mutex_.AssertHeld();
+  assert(lock_wal_count_ == 0);
+
   // TODO: plumb Env::IOActivity, Env::IOPriority
   const ReadOptions read_options;
   const WriteOptions write_options;

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);
@@ -887,7 +921,7 @@ TEST_P(DBWALTestWithParam, WALTrashCleanupOnOpen) {
       // before restarting the DB.
       // We have to set this on the 2nd to last file for it to delay deletion
       // on the last file. (Quirk of DeleteScheduler::BackgroundEmptyTrash())
-      options.sst_file_manager->SetDeleteRateBytesPerSecond(1);
+      options.sst_file_manager->SetDeleteRateBytesPerSecond(1024 * 1024);
     }
     ASSERT_OK(Put("Key2", DummyString(1024, v)));
     ASSERT_OK(Put("Key3", DummyString(1024, v)));
@@ -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);

db/db_write_test.cc

@@ -666,11 +666,25 @@ TEST_P(DBWriteTest, LockWALInEffect) {
   // try the 1st WAL created during open
   ASSERT_OK(Put("key0", "value"));
   ASSERT_NE(options.manual_wal_flush, dbfull()->WALBufferIsEmpty());
+
   ASSERT_OK(db_->LockWAL());
+
   ASSERT_TRUE(dbfull()->WALBufferIsEmpty());
+  uint64_t wal_num = dbfull()->TEST_GetCurrentLogNumber();
+  // Manual flush with wait=false should abruptly fail with TryAgain
+  FlushOptions flush_opts;
+  flush_opts.wait = false;
+  for (bool allow_write_stall : {true, false}) {
+    flush_opts.allow_write_stall = allow_write_stall;
+    ASSERT_TRUE(db_->Flush(flush_opts).IsTryAgain());
+  }
+  ASSERT_EQ(wal_num, dbfull()->TEST_GetCurrentLogNumber());
+
   ASSERT_OK(db_->UnlockWAL());
-  // try the 2nd wal created during SwitchWAL
+
+  // try the 2nd wal created during SwitchWAL (not locked this time)
   ASSERT_OK(dbfull()->TEST_SwitchWAL());
+  ASSERT_NE(wal_num, dbfull()->TEST_GetCurrentLogNumber());
   ASSERT_OK(Put("key1", "value"));
   ASSERT_NE(options.manual_wal_flush, dbfull()->WALBufferIsEmpty());
   ASSERT_OK(db_->LockWAL());
@@ -709,21 +723,57 @@ TEST_P(DBWriteTest, LockWALInEffect) {
 }
 
 TEST_P(DBWriteTest, LockWALConcurrentRecursive) {
+  // This is a micro-stress test of LockWAL and concurrency handling.
+  // It is considered the most convenient way to balance functional
+  // coverage and reproducibility (vs. the two extremes of (a) unit tests
+  // tailored to specific interleavings and (b) db_stress)
   Options options = GetOptions();
   Reopen(options);
-  ASSERT_OK(Put("k1", "val"));
+  ASSERT_OK(Put("k1", "k1_orig"));
   ASSERT_OK(db_->LockWAL());  // 0 -> 1
   auto frozen_seqno = db_->GetLatestSequenceNumber();
-  std::atomic<bool> t1_completed{false};
-  port::Thread t1{[&]() {
-    // Won't finish until WAL unlocked
-    ASSERT_OK(Put("k1", "val2"));
-    t1_completed = true;
+
+  std::string ingest_file = dbname_ + "/external.sst";
+  {
+    SstFileWriter sst_file_writer(EnvOptions(), options);
+    ASSERT_OK(sst_file_writer.Open(ingest_file));
+    ASSERT_OK(sst_file_writer.Put("k2", "k2_val"));
+    ExternalSstFileInfo external_info;
+    ASSERT_OK(sst_file_writer.Finish(&external_info));
+  }
+  AcqRelAtomic<bool> parallel_ingest_completed{false};
+  port::Thread parallel_ingest{[&]() {
+    IngestExternalFileOptions ingest_opts;
+    ingest_opts.move_files = true;  // faster than copy
+    // Shouldn't finish until WAL unlocked
+    ASSERT_OK(db_->IngestExternalFile({ingest_file}, ingest_opts));
+    parallel_ingest_completed.Store(true);
+  }};
+
+  AcqRelAtomic<bool> flush_completed{false};
+  port::Thread parallel_flush{[&]() {
+    FlushOptions flush_opts;
+    // NB: Flush with wait=false case is tested above in LockWALInEffect
+    flush_opts.wait = true;
+    // allow_write_stall = true blocks in fewer cases
+    flush_opts.allow_write_stall = true;
+    // Shouldn't finish until WAL unlocked
+    ASSERT_OK(db_->Flush(flush_opts));
+    flush_completed.Store(true);
+  }};
+
+  AcqRelAtomic<bool> parallel_put_completed{false};
+  port::Thread parallel_put{[&]() {
+    // This can make certain failure scenarios more likely:
+    //   sleep(1);
+    // Shouldn't finish until WAL unlocked
+    ASSERT_OK(Put("k1", "k1_mod"));
+    parallel_put_completed.Store(true);
   }};
 
   ASSERT_OK(db_->LockWAL());  // 1 -> 2
   // Read-only ops are OK
-  ASSERT_EQ(Get("k1"), "val");
+  ASSERT_EQ(Get("k1"), "k1_orig");
   {
     std::vector<LiveFileStorageInfo> files;
     LiveFilesStorageInfoOptions lf_opts;
@@ -732,29 +782,35 @@ TEST_P(DBWriteTest, LockWALConcurrentRecursive) {
     ASSERT_OK(db_->GetLiveFilesStorageInfo({lf_opts}, &files));
   }
 
-  port::Thread t2{[&]() {
+  port::Thread parallel_lock_wal{[&]() {
     ASSERT_OK(db_->LockWAL());  // 2 -> 3 or 1 -> 2
   }};
 
   ASSERT_OK(db_->UnlockWAL());  // 2 -> 1 or 3 -> 2
-  // Give t1 an extra chance to jump in case of bug
+  // Give parallel_put an extra chance to jump in case of bug
   std::this_thread::yield();
-  t2.join();
-  ASSERT_FALSE(t1_completed.load());
+  parallel_lock_wal.join();
+  ASSERT_FALSE(parallel_put_completed.Load());
+  ASSERT_FALSE(parallel_ingest_completed.Load());
+  ASSERT_FALSE(flush_completed.Load());
 
   // Should now have 2 outstanding LockWAL
-  ASSERT_EQ(Get("k1"), "val");
+  ASSERT_EQ(Get("k1"), "k1_orig");
 
   ASSERT_OK(db_->UnlockWAL());  // 2 -> 1
 
-  ASSERT_FALSE(t1_completed.load());
-  ASSERT_EQ(Get("k1"), "val");
+  ASSERT_FALSE(parallel_put_completed.Load());
+  ASSERT_FALSE(parallel_ingest_completed.Load());
+  ASSERT_FALSE(flush_completed.Load());
+
+  ASSERT_EQ(Get("k1"), "k1_orig");
+  ASSERT_EQ(Get("k2"), "NOT_FOUND");
   ASSERT_EQ(frozen_seqno, db_->GetLatestSequenceNumber());
 
   // Ensure final Unlock is concurrency safe and extra Unlock is safe but
   // non-OK
   std::atomic<int> unlock_ok{0};
-  port::Thread t3{[&]() {
+  port::Thread parallel_stuff{[&]() {
     if (db_->UnlockWAL().ok()) {
       unlock_ok++;
     }
@@ -767,18 +823,23 @@ TEST_P(DBWriteTest, LockWALConcurrentRecursive) {
   if (db_->UnlockWAL().ok()) {
     unlock_ok++;
   }
-  t3.join();
+  parallel_stuff.join();
 
   // There was one extra unlock, so just one non-ok
   ASSERT_EQ(unlock_ok.load(), 2);
 
   // Write can proceed
-  t1.join();
-  ASSERT_TRUE(t1_completed.load());
-  ASSERT_EQ(Get("k1"), "val2");
+  parallel_put.join();
+  ASSERT_TRUE(parallel_put_completed.Load());
+  ASSERT_EQ(Get("k1"), "k1_mod");
+  parallel_ingest.join();
+  ASSERT_TRUE(parallel_ingest_completed.Load());
+  ASSERT_EQ(Get("k2"), "k2_val");
+  parallel_flush.join();
+  ASSERT_TRUE(flush_completed.Load());
   // And new writes
-  ASSERT_OK(Put("k2", "val"));
-  ASSERT_EQ(Get("k2"), "val");
+  ASSERT_OK(Put("k3", "val"));
+  ASSERT_EQ(Get("k3"), "val");
 }
 
 TEST_P(DBWriteTest, ConcurrentlyDisabledWAL) {

db/external_sst_file_basic_test.cc

@@ -674,10 +674,6 @@ TEST_F(ExternalSSTFileBasicTest, NoCopy) {
   ASSERT_EQ(file3_info.smallest_key, Key(110));
   ASSERT_EQ(file3_info.largest_key, Key(124));
 
-  ASSERT_OK(dbfull()->LockWAL());
-  // TODO(FIXME): should not allow file ingestion.
-  // With below line, ingestion will block, without it, ingestion can go
-  // through. ASSERT_OK(dbfull()->Put(WriteOptions(), "vo", "vo"));
   s = DeprecatedAddFile({file1}, true /* move file */);
   ASSERT_OK(s) << s.ToString();
   ASSERT_EQ(Status::NotFound(), env_->FileExists(file1));

db/log_reader.cc

@@ -487,9 +487,11 @@ unsigned int Reader::ReadPhysicalRecord(Slice* result, size_t* drop_size,
          type == kRecyclableUserDefinedTimestampSizeType);
     if (is_recyclable_type) {
       header_size = kRecyclableHeaderSize;
-      if (end_of_buffer_offset_ - buffer_.size() == 0) {
-        recycled_ = true;
+      if (first_record_read_ && !recycled_) {
+        // A recycled log should have started with a recycled record
+        return kBadRecord;
       }
+      recycled_ = true;
       // We need enough for the larger header
       if (buffer_.size() < static_cast<size_t>(kRecyclableHeaderSize)) {
         int r = kEof;
@@ -867,9 +869,12 @@ bool FragmentBufferedReader::TryReadFragment(
   int header_size = kHeaderSize;
   if ((type >= kRecyclableFullType && type <= kRecyclableLastType) ||
       type == kRecyclableUserDefinedTimestampSizeType) {
-    if (end_of_buffer_offset_ - buffer_.size() == 0) {
-      recycled_ = true;
+    if (first_record_read_ && !recycled_) {
+      // A recycled log should have started with a recycled record
+      *fragment_type_or_err = kBadRecord;
+      return true;
     }
+    recycled_ = true;
     header_size = kRecyclableHeaderSize;
     while (buffer_.size() < static_cast<size_t>(kRecyclableHeaderSize)) {
       size_t old_size = buffer_.size();

db/log_test.cc

@@ -1157,6 +1157,42 @@ TEST_P(CompressionLogTest, AlignedFragmentation) {
   ASSERT_EQ("EOF", Read());
 }
 
+TEST_P(CompressionLogTest, ChecksumMismatch) {
+  const CompressionType kCompressionType = std::get<2>(GetParam());
+  const bool kCompressionEnabled = kCompressionType != kNoCompression;
+  const bool kRecyclableLog = (std::get<0>(GetParam()) != 0);
+  if (!StreamingCompressionTypeSupported(kCompressionType)) {
+    ROCKSDB_GTEST_SKIP("Test requires support for compression type");
+    return;
+  }
+  ASSERT_OK(SetupTestEnv());
+
+  Write("foooooo");
+  int header_len;
+  if (kRecyclableLog) {
+    header_len = kRecyclableHeaderSize;
+  } else {
+    header_len = kHeaderSize;
+  }
+  int compression_record_len;
+  if (kCompressionEnabled) {
+    compression_record_len = header_len + 4;
+  } else {
+    compression_record_len = 0;
+  }
+  IncrementByte(compression_record_len + header_len /* offset */,
+                14 /* delta */);
+
+  ASSERT_EQ("EOF", Read());
+  if (!kRecyclableLog) {
+    ASSERT_GT(DroppedBytes(), 0U);
+    ASSERT_EQ("OK", MatchError("checksum mismatch"));
+  } else {
+    ASSERT_EQ(0U, DroppedBytes());
+    ASSERT_EQ("", ReportMessage());
+  }
+}
+
 INSTANTIATE_TEST_CASE_P(
     Compression, CompressionLogTest,
     ::testing::Combine(::testing::Values(0, 1), ::testing::Bool(),

db_stress_tool/db_stress_test_base.cc

@@ -958,12 +958,48 @@ void StressTest::OperateDb(ThreadState* thread) {
         if (!s.ok()) {
           fprintf(stderr, "LockWAL() failed: %s\n", s.ToString().c_str());
         } else {
+          // Verify no writes during LockWAL
           auto old_seqno = db_->GetLatestSequenceNumber();
-          // Yield for a while
-          do {
-            std::this_thread::yield();
-          } while (thread->rand.OneIn(2));
-          // Latest seqno should not have changed
+          // And also that WAL is not changed during LockWAL()
+          std::unique_ptr<LogFile> old_wal;
+          s = db_->GetCurrentWalFile(&old_wal);
+          if (!s.ok()) {
+            fprintf(stderr, "GetCurrentWalFile() failed: %s\n",
+                    s.ToString().c_str());
+          } else {
+            // Yield for a while
+            do {
+              std::this_thread::yield();
+            } while (thread->rand.OneIn(2));
+            // Current WAL and size should not have changed
+            std::unique_ptr<LogFile> new_wal;
+            s = db_->GetCurrentWalFile(&new_wal);
+            if (!s.ok()) {
+              fprintf(stderr, "GetCurrentWalFile() failed: %s\n",
+                      s.ToString().c_str());
+            } else {
+              if (old_wal->LogNumber() != new_wal->LogNumber()) {
+                fprintf(stderr,
+                        "Failed: WAL number changed during LockWAL(): %" PRIu64
+                        " to %" PRIu64 "\n",
+                        old_wal->LogNumber(), new_wal->LogNumber());
+              }
+              // FIXME: FaultInjectionTestFS does not report file sizes that
+              // reflect what has been flushed. Either that needs to be fixed
+              // or GetSortedWals/GetLiveWalFile need to stop relying on
+              // asking the FS for sizes.
+              if (!fault_fs_guard &&
+                  old_wal->SizeFileBytes() != new_wal->SizeFileBytes()) {
+                fprintf(stderr,
+                        "Failed: WAL %" PRIu64
+                        " size changed during LockWAL(): %" PRIu64
+                        " to %" PRIu64 "\n",
+                        old_wal->LogNumber(), old_wal->SizeFileBytes(),
+                        new_wal->SizeFileBytes());
+              }
+            }
+          }
+          // Verify no writes during LockWAL
           auto new_seqno = db_->GetLatestSequenceNumber();
           if (old_seqno != new_seqno) {
             fprintf(
@@ -971,6 +1007,7 @@ void StressTest::OperateDb(ThreadState* thread) {
                 "Failure: latest seqno changed from %u to %u with WAL locked\n",
                 (unsigned)old_seqno, (unsigned)new_seqno);
           }
+          // Verification done. Now unlock WAL
           s = db_->UnlockWAL();
           if (!s.ok()) {
             fprintf(stderr, "UnlockWAL() failed: %s\n", s.ToString().c_str());
@@ -2468,6 +2505,7 @@ void StressTest::TestPromoteL0(ThreadState* thread,
 
 Status StressTest::TestFlush(const std::vector<int>& rand_column_families) {
   FlushOptions flush_opts;
+  assert(flush_opts.wait);
   if (FLAGS_atomic_flush) {
     return db_->Flush(flush_opts, column_families_);
   }

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

@@ -122,8 +122,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 {
@@ -131,6 +131,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/db.h

@@ -1679,8 +1679,8 @@ class DB {
   // Freezes the logical state of the DB (by stopping writes), and if WAL is
   // enabled, ensures that state has been flushed to DB files (as in
   // FlushWAL()). This can be used for taking a Checkpoint at a known DB
-  // state, though the user must use options to insure no DB flush is invoked
-  // in this frozen state. Other operations allowed on a "read only" DB should
+  // state, though while the WAL is locked, flushes as part of CreateCheckpoint
+  // and simiar are skipped. Other operations allowed on a "read only" DB should
   // work while frozen. Each LockWAL() call that returns OK must eventually be
   // followed by a corresponding call to UnlockWAL(). Where supported, non-OK
   // status is generally only possible with some kind of corruption or I/O

include/rocksdb/version.h

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

table/block_fetcher.cc

@@ -241,7 +241,7 @@ inline void BlockFetcher::GetBlockContents() {
 // Read a block from the file and verify its checksum. Upon return, io_status_
 // will be updated with the status of the read, and slice_ will be updated
 // with a pointer to the data.
-void BlockFetcher::ReadBlock(bool retry, FSAllocationPtr& fs_buf) {
+void BlockFetcher::ReadBlock(bool retry) {
   FSReadRequest read_req;
   IOOptions opts;
   io_status_ = file_->PrepareIOOptions(read_options_, opts);
@@ -336,7 +336,7 @@ void BlockFetcher::ReadBlock(bool retry, FSAllocationPtr& fs_buf) {
 
   if (io_status_.ok()) {
     InsertCompressedBlockToPersistentCacheIfNeeded();
-    fs_buf = std::move(read_req.fs_scratch);
+    fs_buf_ = std::move(read_req.fs_scratch);
   } else {
     ReleaseFileSystemProvidedBuffer(&read_req);
     direct_io_buf_.reset();
@@ -347,7 +347,6 @@ void BlockFetcher::ReadBlock(bool retry, FSAllocationPtr& fs_buf) {
 }
 
 IOStatus BlockFetcher::ReadBlockContents() {
-  FSAllocationPtr fs_buf;
   if (TryGetUncompressBlockFromPersistentCache()) {
     compression_type_ = kNoCompression;
 #ifndef NDEBUG
@@ -360,15 +359,15 @@ IOStatus BlockFetcher::ReadBlockContents() {
       return io_status_;
     }
   } else if (!TryGetSerializedBlockFromPersistentCache()) {
-    ReadBlock(/*retry =*/false, fs_buf);
+    ReadBlock(/*retry =*/false);
     // If the file system supports retry after corruption, then try to
     // re-read the block and see if it succeeds.
     if (io_status_.IsCorruption() && retry_corrupt_read_) {
-      assert(!fs_buf);
-      ReadBlock(/*retry=*/true, fs_buf);
+      assert(!fs_buf_);
+      ReadBlock(/*retry=*/true);
     }
     if (!io_status_.ok()) {
-      assert(!fs_buf);
+      assert(!fs_buf_);
       return io_status_;
     }
   }
@@ -417,16 +416,15 @@ IOStatus BlockFetcher::ReadAsyncBlockContents() {
         return io_s;
       }
       if (io_s.ok()) {
-        FSAllocationPtr fs_buf;
         // Data Block is already in prefetch.
         got_from_prefetch_buffer_ = true;
         ProcessTrailerIfPresent();
         if (io_status_.IsCorruption() && retry_corrupt_read_) {
           got_from_prefetch_buffer_ = false;
-          ReadBlock(/*retry = */ true, fs_buf);
+          ReadBlock(/*retry = */ true);
         }
         if (!io_status_.ok()) {
-          assert(!fs_buf);
+          assert(!fs_buf_);
           return io_status_;
         }
         used_buf_ = const_cast<char*>(slice_.data());

table/block_fetcher.h

@@ -137,6 +137,7 @@ class BlockFetcher {
   bool for_compaction_ = false;
   bool use_fs_scratch_ = false;
   bool retry_corrupt_read_ = false;
+  FSAllocationPtr fs_buf_;
 
   // return true if found
   bool TryGetUncompressBlockFromPersistentCache();
@@ -152,7 +153,7 @@ class BlockFetcher {
   void InsertCompressedBlockToPersistentCacheIfNeeded();
   void InsertUncompressedBlockToPersistentCacheIfNeeded();
   void ProcessTrailerIfPresent();
-  void ReadBlock(bool retry, FSAllocationPtr& fs_buf);
+  void ReadBlock(bool retry);
 
   void ReleaseFileSystemProvidedBuffer(FSReadRequest* read_req) {
     if (use_fs_scratch_) {

tools/db_crashtest.py

@@ -864,12 +864,6 @@ def finalize_and_sanitize(src_params):
     elif (dest_params.get("use_put_entity_one_in") > 1 and
         dest_params.get("use_timed_put_one_in") == 1):
         dest_params["use_timed_put_one_in"] = 3
-    # TODO: re-enable this combination.
-    if dest_params.get("lock_wal_one_in") != 0 and dest_params["ingest_external_file_one_in"] != 0:
-        if random.choice([0, 1]) == 0:
-            dest_params["ingest_external_file_one_in"] = 0
-        else:
-            dest_params["lock_wal_one_in"] = 0
     return dest_params
 
 def gen_cmd_params(args):

unreleased_history/behavior_changes/fifo-compact-range.md

@@ -1 +0,0 @@
-* CompactRange() with change_level=true on a CF with FIFO compaction will return Status::NotSupported(). 

unreleased_history/behavior_changes/fifo-ingestion.md

@@ -1 +0,0 @@
-* External file ingestion with FIFO compaction will always ingest to L0.

unreleased_history/bug_fixes/.md

@@ -1 +0,0 @@
-Fix a bug where `CompactFiles()` can compact files of range conflict with other ongoing compactions' when `preclude_last_level_data_seconds > 0` is used

unreleased_history/bug_fixes/2pc_wal_sync.md

@@ -1 +0,0 @@
-* Fixed a bug for databases using `DBOptions::allow_2pc == true` (all `TransactionDB`s except `OptimisticTransactionDB`) that have exactly one column family. Due to a missing WAL sync, attempting to open the DB could have returned a `Status::Corruption` with a message like "SST file is ahead of WALs".

unreleased_history/bug_fixes/fix-import-epoch.md

@@ -1 +0,0 @@
-* Fix a bug in CreateColumnFamilyWithImport() where if multiple CFs are imported, we were not resetting files' epoch number and L0 files can have overlapping key range but the same epoch number.

unreleased_history/bug_fixes/inplace_update_race.md

@@ -1 +0,0 @@
-* Fixed race conditions when `ColumnFamilyOptions::inplace_update_support == true` between user overwrites and reads on the same key.

unreleased_history/new_features/get_entity_txn.md

@@ -1 +0,0 @@
-Optimistic transactions and pessimistic transactions with the WriteCommitted policy now support the `GetEntity` API.

unreleased_history/new_features/iterator_property_value_pinned.md

@@ -1 +0,0 @@
-* Added new `Iterator` property, "rocksdb.iterator.is-value-pinned", for checking whether the `Slice` returned by `Iterator::value()` can be used until the `Iterator` is destroyed.

unreleased_history/new_features/multi_get_entity_txn.md

@@ -1 +0,0 @@
-Optimistic transactions and WriteCommitted pessimistic transactions now support the `MultiGetEntity` API.

unreleased_history/new_features/put_entity_txn.md

@@ -1 +0,0 @@
-Optimistic transactions and pessimistic transactions with the WriteCommitted policy now support the `PutEntity` API. Support for read APIs and other write policies (WritePrepared, WriteUnprepared) will be added later.

unreleased_history/public_api_changes/expose_block_based_metadata_cache_options_via_c_api.md

@@ -1 +0,0 @@
-Exposed block based metadata cache options via C API

unreleased_history/public_api_changes/expose_compaction_pri_via_c_api.md

@@ -1 +0,0 @@
-Exposed compaction pri via c api.

unreleased_history/public_api_changes/sec_cache_allow_all.md

@@ -1 +0,0 @@
-Add a kAdmPolicyAllowAll option to TieredAdmissionPolicy that admits all blocks evicted from the primary block cache into the compressed secondary cache.

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

@@ -23,6 +23,8 @@
 #include "port/stack_trace.h"
 #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"
@@ -253,6 +255,13 @@ class CheckpointTest : public testing::Test {
     }
     return result;
   }
+
+  int NumTableFilesAtLevel(int level) {
+    std::string property;
+    EXPECT_TRUE(db_->GetProperty(
+        "rocksdb.num-files-at-level" + std::to_string(level), &property));
+    return atoi(property.c_str());
+  }
 };
 
 TEST_F(CheckpointTest, GetSnapshotLink) {
@@ -975,6 +984,100 @@ TEST_F(CheckpointTest, PutRaceWithCheckpointTrackedWalSync) {
   Close();
 }
 
+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

@@ -1171,14 +1171,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_;