Fix the ZStd checksum (#12005 )

Summary: Somehow we had the wrong checksum when validating the ZStd 1.5.5 download for RocksJava in the previous Pull Request - https://github.com/facebook/rocksdb/pull/9304. This PR fixes that. Pull Request resolved: https://github.com/facebook/rocksdb/pull/12005 Reviewed By: jaykorean Differential Revision: D50840338 Pulled By: cbi42 fbshipit-source-id: 8a92779d3bef013d812eecb89aaaf33fc73991ec
Update HISTORY.md and version for 8.7.3
2026-07-07 14:47:40 +08:00 · 2023-10-31 13:22:47 -07:00 · 2023-10-30 14:07:10 -07:00 · 2023-10-30 11:43:56 -07:00 · 2023-10-25 21:20:16 -07:00 · 2023-10-25 20:48:44 -07:00
85 changed files with 1226 additions and 344 deletions
@@ -206,6 +206,9 @@ executors:
      #  $ docker run --cap-add=SYS_PTRACE --security-opt seccomp=unconfined -it zjay437/rocksdb:0.5 bash
      # option `--cap-add=SYS_PTRACE --security-opt seccomp=unconfined` is used to enable gdb to attach an existing process
      - image: zjay437/rocksdb:0.6
+  linux-java-docker:
+    docker:
+      - image: evolvedbinary/rocksjava:centos6_x64-be

 jobs:
  build-macos:
@@ -590,7 +593,7 @@ jobs:
      - post-steps

  build-linux-java-static:
-    executor: linux-docker
+    executor: linux-java-docker
    resource_class: large
    steps:
      - pre-steps
@@ -603,7 +606,7 @@ jobs:
            which javac && javac -version
      - run:
          name: "Build RocksDBJava Static Library"
-          command: make V=1 J=8 -j8 rocksdbjavastatic
+          command: scl enable devtoolset-7 'make V=1 J=8 -j8 rocksdbjavastatic'
      - post-steps

  build-macos-java:
@@ -1,6 +1,70 @@
 # Rocksdb Change Log
 > NOTE: Entries for next release do not go here. Follow instructions in `unreleased_history/README.txt`

+## 8.7.3 (10/30/2023)
+### Behavior Changes
+* Deleting stale files upon recovery are delegated to SstFileManger if available so they can be rate limited.
+
+## 8.7.2 (10/25/2023)
+### Public API Changes
+* Add new Cache APIs GetSecondaryCacheCapacity() and GetSecondaryCachePinnedUsage() to return the configured capacity, and cache reservation charged to the secondary cache.
+
+### Bug Fixes
+* Fixed a possible underflow when computing the compressed secondary cache share of memory reservations while updating the compressed secondary to total block cache ratio.
+* Fix an assertion failure when UpdeteTieredCache() is called in an idempotent manner.
+
+## 8.7.1 (10/20/2023)
+### Bug Fixes
+* Fix a bug in auto_readahead_size where first_internal_key of index blocks wasn't copied properly resulting in corruption error when first_internal_key was used for comparison.
+* Add bounds check in WBWIIteratorImpl and make BaseDeltaIterator, WriteUnpreparedTxn and WritePreparedTxn respect the upper bound and lower bound in ReadOption. See 11680.
+
+## 8.7.0 (09/22/2023)
+### New Features
+* Added an experimental new "automatic" variant of HyperClockCache that does not require a prior estimate of the average size of cache entries. This variant is activated when HyperClockCacheOptions::estimated\_entry\_charge = 0 and has essentially the same concurrency benefits as the existing HyperClockCache.
+* Add a new statistic `COMPACTION_CPU_TOTAL_TIME` that records cumulative compaction cpu time. This ticker is updated regularly while a compaction is running.
+* Add `GetEntity()` API for ReadOnly DB and Secondary DB.
+* Add a new iterator API `Iterator::Refresh(const Snapshot *)` that allows iterator to be refreshed while using the input snapshot to read.
+* Added a new read option `merge_operand_count_threshold`. When the number of merge operands applied during a successful point lookup exceeds this threshold, the query will return a special OK status with a new subcode `kMergeOperandThresholdExceeded`. Applications might use this signal to take action to reduce the number of merge operands for the affected key(s), for example by running a compaction.
+* For `NewRibbonFilterPolicy()`, made the `bloom_before_level` option mutable through the Configurable interface and the SetOptions API, allowing dynamic switching between all-Bloom and all-Ribbon configurations, and configurations in between. See comments on `NewRibbonFilterPolicy()`
+* RocksDB now allows the block cache to be stacked on top of a compressed secondary cache and a non-volatile secondary cache, thus creating a three-tier cache. To set it up, use the `NewTieredCache()` API in rocksdb/cache.h..
+* Added a new wide-column aware full merge API called `FullMergeV3` to `MergeOperator`. `FullMergeV3` supports wide columns both as base value and merge result, which enables the application to perform more general transformations during merges. For backward compatibility, the default implementation implements the earlier logic of applying the merge operation to the default column of any wide-column entities. Specifically, if there is no base value or the base value is a plain key-value, the default implementation falls back to `FullMergeV2`. If the base value is a wide-column entity, the default implementation invokes `FullMergeV2` to perform the merge on the default column, and leaves any other columns unchanged.
+* Add wide column support to ldb commands (scan, dump, idump, dump_wal) and sst_dump tool's scan command
+
+### Public API Changes
+* Expose more information about input files used in table creation (if any) in `CompactionFilter::Context`. See `CompactionFilter::Context::input_start_level`,`CompactionFilter::Context::input_table_properties` for more.
+* `Options::compaction_readahead_size` 's default value is changed from 0 to 2MB.
+* When using LZ4 compression, the `acceleration` parameter is configurable by setting the negated value in `CompressionOptions::level`. For example, `CompressionOptions::level=-10` will set `acceleration=10`
+* The `NewTieredCache` API has been changed to take the total cache capacity (inclusive of both the primary and the compressed secondary cache) and the ratio of total capacity to allocate to the compressed cache. These are specified in `TieredCacheOptions`. Any capacity specified in `LRUCacheOptions`, `HyperClockCacheOptions` and `CompressedSecondaryCacheOptions` is ignored. A new API, `UpdateTieredCache` is provided to dynamically update the total capacity, ratio of compressed cache, and admission policy.
+* The `NewTieredVolatileCache()` API in rocksdb/cache.h has been renamed to `NewTieredCache()`.
+
+### Behavior Changes
+* Compaction read performance will regress when `Options::compaction_readahead_size` is explicitly set to 0
+* Universal size amp compaction will conditionally exclude some of the newest L0 files when selecting input with a small negative impact to size amp. This is to prevent a large number of L0 files from being locked by a size amp compaction, potentially leading to write stop with a few more flushes.
+* Change ldb scan command delimiter from ':' to '==>'.
+* For non direct IO, eliminate the file system prefetching attempt for compaction read when `Options::compaction_readahead_size` is 0
+
+### Bug Fixes
+* Fix a bug where if there is an error reading from offset 0 of a file from L1+ and that the file is not the first file in the sorted run, data can be lost in compaction and read/scan can return incorrect results.
+* Fix a bug where iterator may return incorrect result for DeleteRange() users if there was an error reading from a file.
+* Fix a bug with atomic_flush=true that can cause DB to stuck after a flush fails (#11872).
+* Fix a bug where RocksDB (with atomic_flush=false) can delete output SST files of pending flushes when a previous concurrent flush fails (#11865). This can result in DB entering read-only state with error message like `IO error: No such file or directory: While open a file for random read: /tmp/rocksdbtest-501/db_flush_test_87732_4230653031040984171/000013.sst`.
+* Fix an assertion fault during seek with async_io when readahead trimming is enabled.
+* When the compressed secondary cache capacity is reduced to 0, it should be completely disabled. Before this fix, inserts and lookups would still go to the backing `LRUCache` before returning, thus incurring locking overhead. With this fix, inserts and lookups are no-ops and do not add any overhead.
+* Updating the tiered cache (cache allocated using NewTieredCache()) by calling SetCapacity() on it was not working properly. The initial creation would set the primary cache capacity to the combined primary and compressed secondary cache capacity. But SetCapacity() would just set the primary cache capacity. With this fix, the user always specifies the total budget and compressed secondary cache ratio on creation. Subsequently, SetCapacity() will distribute the new capacity across the two caches by the same ratio.
+* Fixed a bug in `MultiGet` for cleaning up SuperVersion acquired with locking db mutex.
+* Fix a bug where row cache can falsely return kNotFound even though row cache entry is hit.
+* Fixed a race condition in `GenericRateLimiter` that could cause it to stop granting requests
+* Fix a bug (Issue #10257) where DB can hang after write stall since no compaction is scheduled (#11764).
+* Add a fix for async_io where during seek, when reading a block for seeking a target key in a file without any readahead, the iterator aligned the read on a page boundary and reading more than necessary. This increased the storage read bandwidth usage.
+* Fix an issue in sst dump tool to handle bounds specified for data with user-defined timestamps.
+* When auto_readahead_size is enabled, update readahead upper bound during readahead trimming when reseek changes iterate_upper_bound dynamically.
+* Fixed a bug where `rocksdb.file.read.verify.file.checksums.micros` is not populated
+* Fixed a bug where compaction read under non direct IO still falls back to RocksDB internal prefetching after file system's prefetching returns non-OK status other than `Status::NotSupported()`
+
+### Performance Improvements
+* Added additional improvements in tuning readahead_size during Scans when auto_readahead_size is enabled. However it's not recommended for backward scans and might impact the performance. More details in options.h.
+* During async_io, the Seek happens in 2 phases. Phase 1 starts an asynchronous read on a block cache miss, and phase 2 waits for it to complete and finishes the seek. In both phases, it tries to lookup the block cache for the data block first before looking in the prefetch buffer. It's optimized by doing the block cache lookup only in the first phase that would save some CPU.
+
 ## 8.6.0 (08/18/2023)
 ### New Features
 * Added enhanced data integrity checking on SST files with new format_version=6. Performance impact is very small or negligible. Previously if SST data was misplaced or re-arranged by the storage layer, it could pass block checksum with higher than 1 in 4 billion probability. With format_version=6, block checksums depend on what file they are in and location within the file. This way, misplaced SST data is no more likely to pass checksum verification than randomly corrupted data. Also in format_version=6, SST footers are checksum-protected.
@@ -2094,8 +2094,8 @@ ROCKSDB_JAVADOCS_JAR = rocksdbjni-$(ROCKSDB_JAVA_VERSION)-javadoc.jar
 ROCKSDB_SOURCES_JAR = rocksdbjni-$(ROCKSDB_JAVA_VERSION)-sources.jar
 SHA256_CMD = sha256sum

-ZLIB_VER ?= 1.2.13
-ZLIB_SHA256 ?= b3a24de97a8fdbc835b9833169501030b8977031bcb54b3b3ac13740f846ab30
+ZLIB_VER ?= 1.3
+ZLIB_SHA256 ?= ff0ba4c292013dbc27530b3a81e1f9a813cd39de01ca5e0f8bf355702efa593e
 ZLIB_DOWNLOAD_BASE ?= http://zlib.net
 BZIP2_VER ?= 1.0.8
 BZIP2_SHA256 ?= ab5a03176ee106d3f0fa90e381da478ddae405918153cca248e682cd0c4a2269
@@ -2103,11 +2103,11 @@ BZIP2_DOWNLOAD_BASE ?= http://sourceware.org/pub/bzip2
 SNAPPY_VER ?= 1.1.8
 SNAPPY_SHA256 ?= 16b677f07832a612b0836178db7f374e414f94657c138e6993cbfc5dcc58651f
 SNAPPY_DOWNLOAD_BASE ?= https://github.com/google/snappy/archive
-LZ4_VER ?= 1.9.3
-LZ4_SHA256 ?= 030644df4611007ff7dc962d981f390361e6c97a34e5cbc393ddfbe019ffe2c1
+LZ4_VER ?= 1.9.4
+LZ4_SHA256 ?= 0b0e3aa07c8c063ddf40b082bdf7e37a1562bda40a0ff5272957f3e987e0e54b
 LZ4_DOWNLOAD_BASE ?= https://github.com/lz4/lz4/archive
-ZSTD_VER ?= 1.4.9
-ZSTD_SHA256 ?= acf714d98e3db7b876e5b540cbf6dee298f60eb3c0723104f6d3f065cd60d6a8
+ZSTD_VER ?= 1.5.5
+ZSTD_SHA256 ?= 98e9c3d949d1b924e28e01eccb7deed865eefebf25c2f21c702e5cd5b63b85e1
 ZSTD_DOWNLOAD_BASE ?= https://github.com/facebook/zstd/archive
 CURL_SSL_OPTS ?= --tlsv1

@@ -674,7 +674,7 @@ else
  fi

  if [[ "${PLATFORM}" == "OS_MACOSX" ]]; then
-    # For portability compile for macOS 10.14 or newer
+    # For portability compile for macOS 10.14 (2018) or newer
    COMMON_FLAGS="$COMMON_FLAGS -mmacosx-version-min=10.14"
    PLATFORM_LDFLAGS="$PLATFORM_LDFLAGS -mmacosx-version-min=10.14"
    # -mmacosx-version-min must come first here.
@@ -1276,6 +1276,13 @@ TEST_P(CompressedSecCacheTestWithTiered, DynamicUpdateWithReservation) {
  ASSERT_OK(sec_cache->GetCapacity(sec_capacity));
  ASSERT_EQ(sec_capacity, (30 << 20));

+  ASSERT_OK(tiered_cache->GetSecondaryCacheCapacity(sec_capacity));
+  ASSERT_EQ(sec_capacity, 30 << 20);
+  size_t sec_usage;
+  ASSERT_OK(tiered_cache->GetSecondaryCachePinnedUsage(sec_usage));
+  EXPECT_PRED3(CacheUsageWithinBounds, sec_usage, 3 << 20,
+               GetPercent(3 << 20, 1));
+
  ASSERT_OK(UpdateTieredCache(tiered_cache, -1, 0.39));
  EXPECT_PRED3(CacheUsageWithinBounds, GetCache()->GetUsage(), (45 << 20),
               GetPercent(45 << 20, 1));
@@ -1312,6 +1319,27 @@ TEST_P(CompressedSecCacheTestWithTiered, DynamicUpdateWithReservation) {
  ASSERT_OK(cache_res_mgr()->UpdateCacheReservation(0));
 }

+TEST_P(CompressedSecCacheTestWithTiered,
+       DynamicUpdateWithReservationUnderflow) {
+  std::shared_ptr<Cache> tiered_cache = GetTieredCache();
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->LoadDependency(
+      {{"CacheWithSecondaryAdapter::Release:ChargeSecCache1",
+        "CacheWithSecondaryAdapter::UpdateCacheReservationRatio:Begin"},
+       {"CacheWithSecondaryAdapter::UpdateCacheReservationRatio:End",
+        "CacheWithSecondaryAdapter::Release:ChargeSecCache2"}});
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->EnableProcessing();
+
+  port::Thread reserve_release_thread([&]() {
+    EXPECT_EQ(cache_res_mgr()->UpdateCacheReservation(50), Status::OK());
+    EXPECT_EQ(cache_res_mgr()->UpdateCacheReservation(0), Status::OK());
+  });
+  ASSERT_OK(UpdateTieredCache(tiered_cache, 100 << 20, 0.01));
+  reserve_release_thread.join();
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->DisableProcessing();
+
+  ASSERT_OK(UpdateTieredCache(tiered_cache, 100 << 20, 0.3));
+}
+
 INSTANTIATE_TEST_CASE_P(
    CompressedSecCacheTests, CompressedSecCacheTestWithTiered,
    ::testing::Values(
@@ -5,8 +5,11 @@

 #include "cache/secondary_cache_adapter.h"

+#include <atomic>
+
 #include "cache/tiered_secondary_cache.h"
 #include "monitoring/perf_context_imp.h"
+#include "test_util/sync_point.h"
 #include "util/cast_util.h"

 namespace ROCKSDB_NAMESPACE {
@@ -100,7 +103,8 @@ CacheWithSecondaryAdapter::CacheWithSecondaryAdapter(
    // secondary cache is freed from the reservation.
    s = pri_cache_res_->UpdateCacheReservation(sec_capacity);
    assert(s.ok());
-    sec_cache_res_ratio_ = (double)sec_capacity / target_->GetCapacity();
+    sec_cache_res_ratio_.store((double)sec_capacity / target_->GetCapacity(),
+                               std::memory_order_relaxed);
  }
 }

@@ -233,7 +237,8 @@ Status CacheWithSecondaryAdapter::Insert(const Slice& key, ObjectPtr value,
                                         CompressionType type) {
  Status s = target_->Insert(key, value, helper, charge, handle, priority);
  if (s.ok() && value == nullptr && distribute_cache_res_) {
-    size_t sec_charge = static_cast<size_t>(charge * (sec_cache_res_ratio_));
+    size_t sec_charge = static_cast<size_t>(
+        charge * (sec_cache_res_ratio_.load(std::memory_order_relaxed)));
    s = secondary_cache_->Deflate(sec_charge);
    assert(s.ok());
    s = pri_cache_res_->UpdateCacheReservation(sec_charge, /*increase=*/false);
@@ -241,9 +246,10 @@ Status CacheWithSecondaryAdapter::Insert(const Slice& key, ObjectPtr value,
  }
  // Warm up the secondary cache with the compressed block. The secondary
  // cache may choose to ignore it based on the admission policy.
-  if (value != nullptr && !compressed_value.empty()) {
+  if (value != nullptr && !compressed_value.empty() &&
+      adm_policy_ == TieredAdmissionPolicy::kAdmPolicyThreeQueue) {
    Status status = secondary_cache_->InsertSaved(key, compressed_value, type);
-    assert(status.ok());
+    assert(status.ok() || status.IsNotSupported());
  }

  return s;
@@ -281,7 +287,10 @@ bool CacheWithSecondaryAdapter::Release(Handle* handle,
    ObjectPtr v = target_->Value(handle);
    if (v == nullptr && distribute_cache_res_) {
      size_t charge = target_->GetCharge(handle);
-      size_t sec_charge = static_cast<size_t>(charge * (sec_cache_res_ratio_));
+      size_t sec_charge = static_cast<size_t>(
+          charge * (sec_cache_res_ratio_.load(std::memory_order_relaxed)));
+      TEST_SYNC_POINT("CacheWithSecondaryAdapter::Release:ChargeSecCache1");
+      TEST_SYNC_POINT("CacheWithSecondaryAdapter::Release:ChargeSecCache2");
      Status s = secondary_cache_->Inflate(sec_charge);
      assert(s.ok());
      s = pri_cache_res_->UpdateCacheReservation(sec_charge, /*increase=*/true);
@@ -432,7 +441,9 @@ const char* CacheWithSecondaryAdapter::Name() const {
 // where the new capacity < total cache reservations.
 void CacheWithSecondaryAdapter::SetCapacity(size_t capacity) {
  size_t sec_capacity = static_cast<size_t>(
-      capacity * (distribute_cache_res_ ? sec_cache_res_ratio_ : 0.0));
+      capacity * (distribute_cache_res_
+                      ? sec_cache_res_ratio_.load(std::memory_order_relaxed)
+                      : 0.0));
  size_t old_sec_capacity = 0;

  if (distribute_cache_res_) {
@@ -478,6 +489,29 @@ void CacheWithSecondaryAdapter::SetCapacity(size_t capacity) {
  }
 }

+Status CacheWithSecondaryAdapter::GetSecondaryCacheCapacity(
+    size_t& size) const {
+  return secondary_cache_->GetCapacity(size);
+}
+
+Status CacheWithSecondaryAdapter::GetSecondaryCachePinnedUsage(
+    size_t& size) const {
+  Status s;
+  if (distribute_cache_res_) {
+    MutexLock m(&mutex_);
+    size_t capacity = 0;
+    s = secondary_cache_->GetCapacity(capacity);
+    if (s.ok()) {
+      size = capacity - pri_cache_res_->GetTotalMemoryUsed();
+    } else {
+      size = 0;
+    }
+  } else {
+    size = 0;
+  }
+  return s;
+}
+
 // Update the secondary/primary allocation ratio (remember, the primary
 // capacity is the total memory budget when distribute_cache_res_ is true).
 // When the ratio changes, we may accumulate some error in the calculations
@@ -492,7 +526,8 @@ void CacheWithSecondaryAdapter::SetCapacity(size_t capacity) {
 // in the future.
 Status CacheWithSecondaryAdapter::UpdateCacheReservationRatio(
    double compressed_secondary_ratio) {
-  if (!distribute_cache_res_ || sec_cache_res_ratio_ == 0.0) {
+  if (!distribute_cache_res_ ||
+      sec_cache_res_ratio_.load(std::memory_order_relaxed) == 0.0) {
    return Status::NotSupported();
  }

@@ -506,14 +541,33 @@ Status CacheWithSecondaryAdapter::UpdateCacheReservationRatio(
    return s;
  }

-  assert(old_sec_capacity >= pri_cache_res_->GetTotalMemoryUsed());
-  size_t old_sec_reserved =
-      old_sec_capacity - pri_cache_res_->GetTotalMemoryUsed();
+  TEST_SYNC_POINT(
+      "CacheWithSecondaryAdapter::UpdateCacheReservationRatio:Begin");
+
+  // There's a possible race condition here. Since the read of pri_cache_res_
+  // memory used (secondary cache usage charged to primary cache), and the
+  // change to sec_cache_res_ratio_ are not guarded by a mutex, its possible
+  // that an Insert/Release in another thread might decrease/increase the
+  // pri_cache_res_ reservation by the wrong amount. This should not be a
+  // problem because updating the sec/pri ratio is a rare operation, and
+  // the worst that can happen is we may over/under charge the secondary
+  // cache usage by a little bit. But we do need to protect against
+  // underflow of old_sec_reserved.
+  // TODO: Make the accounting more accurate by tracking the total memory
+  // reservation on the primary cache. This will also allow us to remove
+  // the restriction of not being able to change the sec/pri ratio from
+  // 0.0 to higher.
+  size_t sec_charge_to_pri = pri_cache_res_->GetTotalMemoryUsed();
+  size_t old_sec_reserved = (old_sec_capacity > sec_charge_to_pri)
+                                ? (old_sec_capacity - sec_charge_to_pri)
+                                : 0;
  // Calculate the new secondary cache reservation
  size_t sec_reserved = static_cast<size_t>(
      old_sec_reserved *
-      (double)(compressed_secondary_ratio / sec_cache_res_ratio_));
-  sec_cache_res_ratio_ = compressed_secondary_ratio;
+      (double)(compressed_secondary_ratio /
+               sec_cache_res_ratio_.load(std::memory_order_relaxed)));
+  sec_cache_res_ratio_.store(compressed_secondary_ratio,
+                             std::memory_order_relaxed);
  if (sec_capacity > old_sec_capacity) {
    // We're increasing the ratio, thus ending up with a larger secondary
    // cache and a smaller usable primary cache capacity. Similar to
@@ -526,7 +580,6 @@ Status CacheWithSecondaryAdapter::UpdateCacheReservationRatio(
    //    cache utilization (increase in capacity - increase in share of cache
    //    reservation)
    // 3. Increase secondary cache capacity
-    assert(sec_reserved > old_sec_reserved || sec_reserved == 0);
    s = secondary_cache_->Deflate(sec_reserved - old_sec_reserved);
    assert(s.ok());
    s = pri_cache_res_->UpdateCacheReservation(
@@ -543,7 +596,6 @@ Status CacheWithSecondaryAdapter::UpdateCacheReservationRatio(
    //    reservations)
    // 3. Inflate the secondary cache to give it back the reduction in its
    //    share of cache reservations
-    assert(old_sec_reserved > sec_reserved || sec_reserved == 0);
    s = secondary_cache_->SetCapacity(sec_capacity);
    if (s.ok()) {
      s = pri_cache_res_->UpdateCacheReservation(
@@ -555,6 +607,7 @@ Status CacheWithSecondaryAdapter::UpdateCacheReservationRatio(
    }
  }

+  TEST_SYNC_POINT("CacheWithSecondaryAdapter::UpdateCacheReservationRatio:End");
 #ifndef NDEBUG
  // As mentioned in the function comments, we may accumulate some erros when
  // the ratio is changed. We set a flag here which disables some assertions
@@ -575,11 +628,40 @@ std::shared_ptr<Cache> NewTieredCache(const TieredCacheOptions& _opts) {
    return nullptr;
  }

-  if (_opts.adm_policy >= TieredAdmissionPolicy::kAdmPolicyMax) {
-    return nullptr;
+  TieredCacheOptions opts = _opts;
+  {
+    bool valid_adm_policy = true;
+
+    switch (_opts.adm_policy) {
+      case TieredAdmissionPolicy::kAdmPolicyAuto:
+        // Select an appropriate default policy
+        if (opts.adm_policy == TieredAdmissionPolicy::kAdmPolicyAuto) {
+          if (opts.nvm_sec_cache) {
+            opts.adm_policy = TieredAdmissionPolicy::kAdmPolicyThreeQueue;
+          } else {
+            opts.adm_policy = TieredAdmissionPolicy::kAdmPolicyPlaceholder;
+          }
+        }
+        break;
+      case TieredAdmissionPolicy::kAdmPolicyPlaceholder:
+      case TieredAdmissionPolicy::kAdmPolicyAllowCacheHits:
+        if (opts.nvm_sec_cache) {
+          valid_adm_policy = false;
+        }
+        break;
+      case TieredAdmissionPolicy::kAdmPolicyThreeQueue:
+        if (!opts.nvm_sec_cache) {
+          valid_adm_policy = false;
+        }
+        break;
+      default:
+        valid_adm_policy = false;
+    }
+    if (!valid_adm_policy) {
+      return nullptr;
+    }
  }

-  TieredCacheOptions opts = _opts;
  std::shared_ptr<Cache> cache;
  if (opts.cache_type == PrimaryCacheType::kCacheTypeLRU) {
    LRUCacheOptions cache_opts =
@@ -602,8 +684,7 @@ std::shared_ptr<Cache> NewTieredCache(const TieredCacheOptions& _opts) {
  sec_cache = NewCompressedSecondaryCache(opts.comp_cache_opts);

  if (opts.nvm_sec_cache) {
-    if (opts.adm_policy == TieredAdmissionPolicy::kAdmPolicyThreeQueue ||
-        opts.adm_policy == TieredAdmissionPolicy::kAdmPolicyAuto) {
+    if (opts.adm_policy == TieredAdmissionPolicy::kAdmPolicyThreeQueue) {
      sec_cache = std::make_shared<TieredSecondaryCache>(
          sec_cache, opts.nvm_sec_cache,
          TieredAdmissionPolicy::kAdmPolicyThreeQueue);
@@ -47,6 +47,10 @@ class CacheWithSecondaryAdapter : public CacheWrapper {

  void SetCapacity(size_t capacity) override;

+  Status GetSecondaryCacheCapacity(size_t& size) const override;
+
+  Status GetSecondaryCachePinnedUsage(size_t& size) const override;
+
  Status UpdateCacheReservationRatio(double ratio);

  Status UpdateAdmissionPolicy(TieredAdmissionPolicy adm_policy);
@@ -80,8 +84,8 @@ class CacheWithSecondaryAdapter : public CacheWrapper {
  std::shared_ptr<ConcurrentCacheReservationManager> pri_cache_res_;
  // Fraction of a cache memory reservation to be assigned to the secondary
  // cache
-  double sec_cache_res_ratio_;
-  port::Mutex mutex_;
+  std::atomic<double> sec_cache_res_ratio_;
+  mutable port::Mutex mutex_;
 #ifndef NDEBUG
  bool ratio_changed_ = false;
 #endif
@@ -83,6 +83,16 @@ size_t ShardedCacheBase::GetCapacity() const {
  return capacity_;
 }

+Status ShardedCacheBase::GetSecondaryCacheCapacity(size_t& size) const {
+  size = 0;
+  return Status::OK();
+}
+
+Status ShardedCacheBase::GetSecondaryCachePinnedUsage(size_t& size) const {
+  size = 0;
+  return Status::OK();
+}
+
 bool ShardedCacheBase::HasStrictCapacityLimit() const {
  MutexLock l(&config_mutex_);
  return strict_capacity_limit_;
@@ -99,6 +99,8 @@ class ShardedCacheBase : public Cache {

  bool HasStrictCapacityLimit() const override;
  size_t GetCapacity() const override;
+  Status GetSecondaryCacheCapacity(size_t& size) const override;
+  Status GetSecondaryCachePinnedUsage(size_t& size) const override;

  using Cache::GetUsage;
  size_t GetUsage(Handle* handle) const override;
@@ -3,6 +3,8 @@
 //  COPYING file in the root directory) and Apache 2.0 License
 //  (found in the LICENSE.Apache file in the root directory).
 //
+#include "cache/compressed_secondary_cache.h"
+#include "cache/secondary_cache_adapter.h"
 #include "db/db_test_util.h"
 #include "rocksdb/cache.h"
 #include "rocksdb/secondary_cache.h"
@@ -179,7 +181,9 @@ class DBTieredSecondaryCacheTest : public DBTestBase {

  std::shared_ptr<Cache> NewCache(size_t pri_capacity,
                                  size_t compressed_capacity,
-                                  size_t nvm_capacity) {
+                                  size_t nvm_capacity,
+                                  TieredAdmissionPolicy adm_policy =
+                                      TieredAdmissionPolicy::kAdmPolicyAuto) {
    LRUCacheOptions lru_opts;
    TieredCacheOptions opts;
    lru_opts.capacity = 0;
@@ -187,30 +191,30 @@ class DBTieredSecondaryCacheTest : public DBTestBase {
    lru_opts.high_pri_pool_ratio = 0;
    opts.cache_opts = &lru_opts;
    opts.cache_type = PrimaryCacheType::kCacheTypeLRU;
-    opts.adm_policy = TieredAdmissionPolicy::kAdmPolicyThreeQueue;
    opts.comp_cache_opts.capacity = 0;
    opts.comp_cache_opts.num_shard_bits = 0;
    opts.total_capacity = pri_capacity + compressed_capacity;
    opts.compressed_secondary_ratio =
        (double)compressed_capacity / opts.total_capacity;
-    nvm_sec_cache_.reset(new TestSecondaryCache(nvm_capacity));
-    opts.nvm_sec_cache = nvm_sec_cache_;
+    if (nvm_capacity > 0) {
+      nvm_sec_cache_.reset(new TestSecondaryCache(nvm_capacity));
+      opts.nvm_sec_cache = nvm_sec_cache_;
+    }
+    opts.adm_policy = adm_policy;
    cache_ = NewTieredCache(opts);
    assert(cache_ != nullptr);

-#if 0
-    CacheWithSecondaryAdapter* adapter_cache_ =
-      static_cast<CacheWithSecondaryAdapter*>(cache_.get());
-    TieredSecondaryCache* tiered_cache_ =
-      static_cast<TieredSecondaryCache*>(
-          adapter_cache_->TEST_GetSecondaryCache());
-#endif
-
    return cache_;
  }

  TestSecondaryCache* nvm_sec_cache() { return nvm_sec_cache_.get(); }

+  CompressedSecondaryCache* compressed_secondary_cache() {
+    return static_cast<CompressedSecondaryCache*>(
+        static_cast<CacheWithSecondaryAdapter*>(cache_.get())
+            ->TEST_GetSecondaryCache());
+  }
+
 private:
  std::shared_ptr<Cache> cache_;
  std::shared_ptr<TestSecondaryCache> nvm_sec_cache_;
@@ -636,6 +640,68 @@ TEST_F(DBTieredSecondaryCacheTest, IterateTest) {
  Destroy(options);
 }

+class DBTieredAdmPolicyTest
+    : public DBTieredSecondaryCacheTest,
+      public testing::WithParamInterface<TieredAdmissionPolicy> {};
+
+TEST_P(DBTieredAdmPolicyTest, CompressedOnlyTest) {
+  if (!LZ4_Supported()) {
+    ROCKSDB_GTEST_SKIP("This test requires LZ4 support.");
+    return;
+  }
+
+  BlockBasedTableOptions table_options;
+  // We want a block cache of size 10KB, and a compressed secondary cache of
+  // size 10KB. However, we specify a block cache size of 256KB here in order
+  // to take into account the cache reservation in the block cache on
+  // behalf of the compressed cache. The unit of cache reservation is 256KB.
+  // The effective block cache capacity will be calculated as 256 + 10 = 266KB,
+  // and 256KB will be reserved for the compressed cache, leaving 10KB for
+  // the primary block cache. We only have to worry about this here because
+  // the cache size is so small.
+  table_options.block_cache = NewCache(256 * 1024, 10 * 1024, 0, GetParam());
+  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));
+
+  size_t comp_cache_usage = compressed_secondary_cache()->TEST_GetUsage();
+  // Disable paranoid_file_checks so that flush will not read back the newly
+  // written file
+  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());
+
+  // The first 2 Gets, for keys 0 and 5, will load the corresponding data
+  // blocks as they will be cache misses. Since this is a 2-tier cache (
+  // primary and compressed), no warm-up should happen with the compressed
+  // blocks.
+  std::string v = Get(Key(0));
+  ASSERT_EQ(1007, v.size());
+
+  v = Get(Key(5));
+  ASSERT_EQ(1007, v.size());
+
+  ASSERT_EQ(compressed_secondary_cache()->TEST_GetUsage(), comp_cache_usage);
+
+  Destroy(options);
+}
+
+INSTANTIATE_TEST_CASE_P(
+    DBTieredAdmPolicyTest, DBTieredAdmPolicyTest,
+    ::testing::Values(TieredAdmissionPolicy::kAdmPolicyAuto,
+                      TieredAdmissionPolicy::kAdmPolicyPlaceholder,
+                      TieredAdmissionPolicy::kAdmPolicyAllowCacheHits));
+
 }  // namespace ROCKSDB_NAMESPACE

 int main(int argc, char** argv) {
@@ -1938,6 +1938,7 @@ Status CompactionJob::OpenCompactionOutputFile(SubcompactionState* sub_compact,
      db_options_.stats, listeners, db_options_.file_checksum_gen_factory.get(),
      tmp_set.Contains(FileType::kTableFile), false));

+  // TODO(hx235): pass in the correct `oldest_key_time` instead of `0`
  TableBuilderOptions tboptions(
      *cfd->ioptions(), *(sub_compact->compaction->mutable_cf_options()),
      cfd->internal_comparator(), cfd->int_tbl_prop_collector_factories(),
@@ -354,7 +354,6 @@ Status DBImpl::ResumeImpl(DBRecoverContext context) {
  }

  // Make sure the IO Status stored in version set is set to OK.
-  bool file_deletion_disabled = !IsFileDeletionsEnabled();
  if (s.ok()) {
    IOStatus io_s = versions_->io_status();
    if (io_s.IsIOError()) {
@@ -363,7 +362,7 @@ Status DBImpl::ResumeImpl(DBRecoverContext context) {
      // clean-up phase MANIFEST writing. We must have also disabled file
      // deletions.
      assert(!versions_->descriptor_log_);
-      assert(file_deletion_disabled);
+      assert(!IsFileDeletionsEnabled());
      // Since we are trying to recover from MANIFEST write error, we need to
      // switch to a new MANIFEST anyway. The old MANIFEST can be corrupted.
      // Therefore, force writing a dummy version edit because we do not know
@@ -401,34 +400,6 @@ Status DBImpl::ResumeImpl(DBRecoverContext context) {
    }
  }

-  JobContext job_context(0);
-  FindObsoleteFiles(&job_context, true);
-  mutex_.Unlock();
-
-  job_context.manifest_file_number = 1;
-  if (job_context.HaveSomethingToDelete()) {
-    PurgeObsoleteFiles(job_context);
-  }
-  job_context.Clean();
-
-  if (s.ok()) {
-    assert(versions_->io_status().ok());
-    // If we reach here, we should re-enable file deletions if it was disabled
-    // during previous error handling.
-    if (file_deletion_disabled) {
-      // Always return ok
-      s = EnableFileDeletions(/*force=*/true);
-      if (!s.ok()) {
-        ROCKS_LOG_INFO(
-            immutable_db_options_.info_log,
-            "DB resume requested but could not enable file deletions [%s]",
-            s.ToString().c_str());
-        assert(false);
-      }
-    }
-  }
-
-  mutex_.Lock();
  if (s.ok()) {
    // This will notify and unblock threads waiting for error recovery to
    // finish. Those previouly waiting threads can now proceed, which may
@@ -441,6 +412,15 @@ Status DBImpl::ResumeImpl(DBRecoverContext context) {
    error_handler_.GetRecoveryError().PermitUncheckedError();
  }

+  JobContext job_context(0);
+  FindObsoleteFiles(&job_context, true);
+  mutex_.Unlock();
+  job_context.manifest_file_number = 1;
+  if (job_context.HaveSomethingToDelete()) {
+    PurgeObsoleteFiles(job_context);
+  }
+  job_context.Clean();
+
  if (s.ok()) {
    ROCKS_LOG_INFO(immutable_db_options_.info_log, "Successfully resumed DB");
  } else {
@@ -448,12 +428,13 @@ Status DBImpl::ResumeImpl(DBRecoverContext context) {
                   s.ToString().c_str());
  }

+  mutex_.Lock();
  // Check for shutdown again before scheduling further compactions,
  // since we released and re-acquired the lock above
  if (shutdown_initiated_) {
    s = Status::ShutdownInProgress();
  }
-  if (s.ok()) {
+  if (s.ok() && context.flush_after_recovery) {
    // Since we drop all non-recovery flush requests during recovery,
    // and new memtable may fill up during recovery,
    // schedule one more round of flush.
@@ -472,14 +453,16 @@ Status DBImpl::ResumeImpl(DBRecoverContext context) {
    // FlushAllColumnFamilies releases and re-acquires mutex.
    if (shutdown_initiated_) {
      s = Status::ShutdownInProgress();
-    } else {
-      for (auto cfd : *versions_->GetColumnFamilySet()) {
-        SchedulePendingCompaction(cfd);
-      }
-      MaybeScheduleFlushOrCompaction();
    }
  }

+  if (s.ok()) {
+    for (auto cfd : *versions_->GetColumnFamilySet()) {
+      SchedulePendingCompaction(cfd);
+    }
+    MaybeScheduleFlushOrCompaction();
+  }
+
  // Wake up any waiters - in this case, it could be the shutdown thread
  bg_cv_.SignalAll();

@@ -4047,7 +4030,8 @@ void DBImpl::ReleaseSnapshot(const Snapshot* s) {
      CfdList cf_scheduled;
      for (auto* cfd : *versions_->GetColumnFamilySet()) {
        if (!cfd->ioptions()->allow_ingest_behind) {
-          cfd->current()->storage_info()->UpdateOldestSnapshot(oldest_snapshot);
+          cfd->current()->storage_info()->UpdateOldestSnapshot(
+              oldest_snapshot, /*allow_ingest_behind=*/false);
          if (!cfd->current()
                   ->storage_info()
                   ->BottommostFilesMarkedForCompaction()
@@ -1383,8 +1383,10 @@ class DBImpl : public DB {
    autovector<ColumnFamilyData*> cfds_;
    autovector<const MutableCFOptions*> mutable_cf_opts_;
    autovector<autovector<VersionEdit*>> edit_lists_;
-    // files_to_delete_ contains sst files
-    std::unordered_set<std::string> files_to_delete_;
+    // Stale SST files to delete found upon recovery. This stores a mapping from
+    // such a file's absolute path to its parent directory.
+    std::unordered_map<std::string, std::string> files_to_delete_;
+    bool is_new_db_ = false;
  };

  // Except in DB::Open(), WriteOptionsFile can only be called when:
@@ -1852,7 +1854,8 @@ class DBImpl : public DB {
  void ReleaseFileNumberFromPendingOutputs(
      std::unique_ptr<std::list<uint64_t>::iterator>& v);

-  IOStatus SyncClosedLogs(JobContext* job_context, VersionEdit* synced_wals);
+  IOStatus SyncClosedLogs(JobContext* job_context, VersionEdit* synced_wals,
+                          bool error_recovery_in_prog);

  // Flush the in-memory write buffer to storage.  Switches to a new
  // log-file/memtable and writes a new descriptor iff successful. Then
@@ -2351,6 +2354,10 @@ class DBImpl : public DB {

  Status DisableFileDeletionsWithLock();

+  // Safely decrease `disable_delete_obsolete_files_` by one while holding lock
+  // and return its remaning value.
+  int EnableFileDeletionsWithLock();
+
  Status IncreaseFullHistoryTsLowImpl(ColumnFamilyData* cfd,
                                      std::string ts_low);

@@ -113,7 +113,8 @@ bool DBImpl::ShouldRescheduleFlushRequestToRetainUDT(
 }

 IOStatus DBImpl::SyncClosedLogs(JobContext* job_context,
-                                VersionEdit* synced_wals) {
+                                VersionEdit* synced_wals,
+                                bool error_recovery_in_prog) {
  TEST_SYNC_POINT("DBImpl::SyncClosedLogs:Start");
  InstrumentedMutexLock l(&log_write_mutex_);
  autovector<log::Writer*, 1> logs_to_sync;
@@ -139,7 +140,7 @@ IOStatus DBImpl::SyncClosedLogs(JobContext* job_context,
      ROCKS_LOG_INFO(immutable_db_options_.info_log,
                     "[JOB %d] Syncing log #%" PRIu64, job_context->job_id,
                     log->get_log_number());
-      if (error_handler_.IsRecoveryInProgress()) {
+      if (error_recovery_in_prog) {
        log->file()->reset_seen_error();
      }
      io_s = log->file()->Sync(immutable_db_options_.use_fsync);
@@ -148,7 +149,7 @@ IOStatus DBImpl::SyncClosedLogs(JobContext* job_context,
      }

      if (immutable_db_options_.recycle_log_file_num > 0) {
-        if (error_handler_.IsRecoveryInProgress()) {
+        if (error_recovery_in_prog) {
          log->file()->reset_seen_error();
        }
        io_s = log->Close();
@@ -261,8 +262,10 @@ Status DBImpl::FlushMemTableToOutputFile(
    // SyncClosedLogs() may unlock and re-lock the log_write_mutex multiple
    // times.
    VersionEdit synced_wals;
+    bool error_recovery_in_prog = error_handler_.IsRecoveryInProgress();
    mutex_.Unlock();
-    log_io_s = SyncClosedLogs(job_context, &synced_wals);
+    log_io_s =
+        SyncClosedLogs(job_context, &synced_wals, error_recovery_in_prog);
    mutex_.Lock();
    if (log_io_s.ok() && synced_wals.IsWalAddition()) {
      const ReadOptions read_options(Env::IOActivity::kFlush);
@@ -546,8 +549,10 @@ Status DBImpl::AtomicFlushMemTablesToOutputFiles(
    // TODO (yanqin) investigate whether we should sync the closed logs for
    // single column family case.
    VersionEdit synced_wals;
+    bool error_recovery_in_prog = error_handler_.IsRecoveryInProgress();
    mutex_.Unlock();
-    log_io_s = SyncClosedLogs(job_context, &synced_wals);
+    log_io_s =
+        SyncClosedLogs(job_context, &synced_wals, error_recovery_in_prog);
    mutex_.Lock();
    if (log_io_s.ok() && synced_wals.IsWalAddition()) {
      const ReadOptions read_options(Env::IOActivity::kFlush);
@@ -100,6 +100,14 @@ Status DBImpl::EnableFileDeletions(bool force) {
  return Status::OK();
 }

+int DBImpl::EnableFileDeletionsWithLock() {
+  mutex_.AssertHeld();
+  // In case others have called EnableFileDeletions(true /* force */) in between
+  disable_delete_obsolete_files_ =
+      std::max(0, disable_delete_obsolete_files_ - 1);
+  return disable_delete_obsolete_files_;
+}
+
 bool DBImpl::IsFileDeletionsEnabled() const {
  return 0 == disable_delete_obsolete_files_;
 }
@@ -995,7 +1003,7 @@ Status DBImpl::DeleteUnreferencedSstFiles(RecoveryContext* recovery_ctx) {
      if (type == kTableFile && number >= next_file_number &&
          recovery_ctx->files_to_delete_.find(normalized_fpath) ==
              recovery_ctx->files_to_delete_.end()) {
-        recovery_ctx->files_to_delete_.emplace(normalized_fpath);
+        recovery_ctx->files_to_delete_.emplace(normalized_fpath, path);
      }
    }
  }
@@ -927,8 +927,11 @@ Status DBImpl::LogAndApplyForRecovery(const RecoveryContext& recovery_ctx) {
      recovery_ctx.edit_lists_, &mutex_, directories_.GetDbDir());
  if (s.ok() && !(recovery_ctx.files_to_delete_.empty())) {
    mutex_.Unlock();
-    for (const auto& fname : recovery_ctx.files_to_delete_) {
-      s = env_->DeleteFile(fname);
+    for (const auto& stale_sst_file : recovery_ctx.files_to_delete_) {
+      s = DeleteDBFile(&immutable_db_options_, stale_sst_file.first,
+                       stale_sst_file.second,
+                       /*force_bg=*/false,
+                       /*force_fg=*/false);
      if (!s.ok()) {
        break;
      }
@@ -83,7 +83,8 @@ DBIter::DBIter(Env* _env, const ReadOptions& read_options,
      cfd_(cfd),
      timestamp_ub_(read_options.timestamp),
      timestamp_lb_(read_options.iter_start_ts),
-      timestamp_size_(timestamp_ub_ ? timestamp_ub_->size() : 0) {
+      timestamp_size_(timestamp_ub_ ? timestamp_ub_->size() : 0),
+      auto_readahead_size_(read_options.auto_readahead_size) {
  RecordTick(statistics_, NO_ITERATOR_CREATED);
  if (pin_thru_lifetime_) {
    pinned_iters_mgr_.StartPinning();
@@ -743,15 +744,22 @@ bool DBIter::ReverseToBackward() {
  // When current_entry_is_merged_ is true, iter_ may be positioned on the next
  // key, which may not exist or may have prefix different from current.
  // If that's the case, seek to saved_key_.
-  if (current_entry_is_merged_ &&
-      (!expect_total_order_inner_iter() || !iter_.Valid())) {
+  //
+  // In case of auto_readahead_size enabled, index_iter moves forward during
+  // forward scan for block cache lookup and points to different block. If Prev
+  // op is called, it needs to call SeekForPrev to point to right index_iter_ in
+  // BlockBasedTableIterator. This only happens when direction is changed from
+  // forward to backward.
+  if ((current_entry_is_merged_ &&
+       (!expect_total_order_inner_iter() || !iter_.Valid())) ||
+      auto_readahead_size_) {
    IterKey last_key;
    // Using kMaxSequenceNumber and kValueTypeForSeek
    // (not kValueTypeForSeekForPrev) to seek to a key strictly smaller
    // than saved_key_.
    last_key.SetInternalKey(ParsedInternalKey(
        saved_key_.GetUserKey(), kMaxSequenceNumber, kValueTypeForSeek));
-    if (!expect_total_order_inner_iter()) {
+    if (!expect_total_order_inner_iter() || auto_readahead_size_) {
      iter_.SeekForPrev(last_key.GetInternalKey());
    } else {
      // Some iterators may not support SeekForPrev(), so we avoid using it
@@ -402,6 +402,7 @@ class DBIter final : public Iterator {
  const Slice* const timestamp_lb_;
  const size_t timestamp_size_;
  std::string saved_timestamp_;
+  bool auto_readahead_size_;
 };

 // Return a new iterator that converts internal keys (yielded by
@@ -937,12 +937,21 @@ INSTANTIATE_TEST_CASE_P(DBWALTestWithParam, DBWALTestWithParam,
                        ::testing::Values(std::make_tuple("", true),
                                          std::make_tuple("_wal_dir", false)));

-TEST_F(DBSSTTest, OpenDBWithExistingTrash) {
+TEST_F(DBSSTTest, OpenDBWithExistingTrashAndObsoleteSstFile) {
  Options options = CurrentOptions();
-
  options.sst_file_manager.reset(
      NewSstFileManager(env_, nullptr, "", 1024 * 1024 /* 1 MB/sec */));
  auto sfm = static_cast<SstFileManagerImpl*>(options.sst_file_manager.get());
+  // Set an extra high trash ratio to prevent immediate/non-rate limited
+  // deletions
+  sfm->SetDeleteRateBytesPerSecond(1024 * 1024);
+  sfm->delete_scheduler()->SetMaxTrashDBRatio(1000.0);
+
+  int bg_delete_file = 0;
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack(
+      "DeleteScheduler::DeleteTrashFile:DeleteFile",
+      [&](void* /*arg*/) { bg_delete_file++; });
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->EnableProcessing();

  Destroy(last_options_);

@@ -951,13 +960,22 @@ TEST_F(DBSSTTest, OpenDBWithExistingTrash) {
  ASSERT_OK(WriteStringToFile(env_, "abc", dbname_ + "/" + "001.sst.trash"));
  ASSERT_OK(WriteStringToFile(env_, "abc", dbname_ + "/" + "002.sst.trash"));
  ASSERT_OK(WriteStringToFile(env_, "abc", dbname_ + "/" + "003.sst.trash"));
+  // Manually add an obsolete sst file. Obsolete SST files are discovered and
+  // deleted upon recovery.
+  constexpr uint64_t kSstFileNumber = 100;
+  const std::string kObsoleteSstFile =
+      MakeTableFileName(dbname_, kSstFileNumber);
+  ASSERT_OK(WriteStringToFile(env_, "abc", kObsoleteSstFile));

-  // Reopen the DB and verify that it deletes existing trash files
+  // Reopen the DB and verify that it deletes existing trash files and obsolete
+  // SST files with rate limiting.
  Reopen(options);
  sfm->WaitForEmptyTrash();
  ASSERT_NOK(env_->FileExists(dbname_ + "/" + "001.sst.trash"));
  ASSERT_NOK(env_->FileExists(dbname_ + "/" + "002.sst.trash"));
  ASSERT_NOK(env_->FileExists(dbname_ + "/" + "003.sst.trash"));
+  ASSERT_NOK(env_->FileExists(kObsoleteSstFile));
+  ASSERT_EQ(bg_delete_file, 4);
 }

 // Create a DB with 2 db_paths, and generate multiple files in the 2
@@ -320,7 +320,7 @@ const Status& ErrorHandler::HandleKnownErrors(const Status& bg_err,
  // Check if recovery is currently in progress. If it is, we will save this
  // error so we can check it at the end to see if recovery succeeded or not
  if (recovery_in_prog_ && recovery_error_.ok()) {
-    recovery_error_ = new_bg_err;
+    recovery_error_ = status_to_io_status(Status(new_bg_err));
  }

  bool auto_recovery = auto_recovery_;
@@ -396,14 +396,13 @@ const Status& ErrorHandler::SetBGError(const Status& bg_status,
  ROCKS_LOG_WARN(db_options_.info_log, "Background IO error %s",
                 bg_io_err.ToString().c_str());

-  if (recovery_in_prog_ && recovery_io_error_.ok()) {
-    recovery_io_error_ = bg_io_err;
-  }
-  if (BackgroundErrorReason::kManifestWrite == reason ||
-      BackgroundErrorReason::kManifestWriteNoWAL == reason) {
+  if (!recovery_disabled_file_deletion_ &&
+      (BackgroundErrorReason::kManifestWrite == reason ||
+       BackgroundErrorReason::kManifestWriteNoWAL == reason)) {
    // Always returns ok
    ROCKS_LOG_INFO(db_options_.info_log, "Disabling File Deletions");
    db_->DisableFileDeletionsWithLock().PermitUncheckedError();
+    recovery_disabled_file_deletion_ = true;
  }

  Status new_bg_io_err = bg_io_err;
@@ -502,10 +501,6 @@ const Status& ErrorHandler::SetBGError(const Status& bg_status,
      RecordTick(bg_error_stats_.get(),
                 ERROR_HANDLER_BG_IO_ERROR_COUNT_MISSPELLED);
    }
-    // HandleKnownErrors() will use recovery_error_, so ignore
-    // recovery_io_error_.
-    // TODO: Do some refactoring and use only one recovery_error_
-    recovery_io_error_.PermitUncheckedError();
    return HandleKnownErrors(new_bg_io_err, reason);
  }
 }
@@ -562,11 +557,23 @@ Status ErrorHandler::ClearBGError() {
    old_bg_error.PermitUncheckedError();
    // Clear and check the recovery IO and BG error
    bg_error_ = Status::OK();
-    recovery_io_error_ = IOStatus::OK();
+    recovery_error_ = IOStatus::OK();
    bg_error_.PermitUncheckedError();
-    recovery_io_error_.PermitUncheckedError();
+    recovery_error_.PermitUncheckedError();
    recovery_in_prog_ = false;
    soft_error_no_bg_work_ = false;
+    if (recovery_disabled_file_deletion_) {
+      recovery_disabled_file_deletion_ = false;
+      int remain_counter = db_->EnableFileDeletionsWithLock();
+      if (remain_counter == 0) {
+        ROCKS_LOG_INFO(db_options_.info_log, "File Deletions Enabled");
+      } else {
+        ROCKS_LOG_WARN(
+            db_options_.info_log,
+            "File Deletions Enable, but not really enabled. Counter: %d",
+            remain_counter);
+      }
+    }
    EventHelpers::NotifyOnErrorRecoveryEnd(db_options_.listeners, old_bg_error,
                                           bg_error_, db_mutex_);
  }
@@ -602,14 +609,14 @@ Status ErrorHandler::RecoverFromBGError(bool is_manual) {
  if (bg_error_.severity() == Status::Severity::kSoftError &&
      recover_context_.flush_reason == FlushReason::kErrorRecovery) {
    // Simply clear the background error and return
-    recovery_error_ = Status::OK();
+    recovery_error_ = IOStatus::OK();
    return ClearBGError();
  }

  // Reset recovery_error_. We will use this to record any errors that happen
  // during the recovery process. While recovering, the only operations that
  // can generate background errors should be the flush operations
-  recovery_error_ = Status::OK();
+  recovery_error_ = IOStatus::OK();
  recovery_error_.PermitUncheckedError();
  Status s = db_->ResumeImpl(recover_context_);
  if (s.ok()) {
@@ -645,21 +652,27 @@ const Status& ErrorHandler::StartRecoverFromRetryableBGIOError(
  ROCKS_LOG_INFO(
      db_options_.info_log,
      "ErrorHandler: Call StartRecoverFromRetryableBGIOError to resume\n");
+  // Needs to be set in the same lock hold as setting BG error, otherwise
+  // intervening writes could see a BG error without a recovery and bail out.
+  recovery_in_prog_ = true;
+
  if (recovery_thread_) {
+    // Ensure only one thread can execute the join().
+    std::unique_ptr<port::Thread> old_recovery_thread(
+        std::move(recovery_thread_));
    // In this case, if recovery_in_prog_ is false, current thread should
    // wait the previous recover thread to finish and create a new thread
    // to recover from the bg error.
    db_mutex_->Unlock();
-    recovery_thread_->join();
+    old_recovery_thread->join();
    db_mutex_->Lock();
  }

-  recovery_in_prog_ = true;
  TEST_SYNC_POINT("StartRecoverFromRetryableBGIOError::in_progress");
  recovery_thread_.reset(
      new port::Thread(&ErrorHandler::RecoverFromRetryableBGIOError, this));

-  if (recovery_io_error_.ok() && recovery_error_.ok()) {
+  if (recovery_error_.ok()) {
    return recovery_error_;
  } else {
    return bg_error_;
@@ -676,9 +689,12 @@ void ErrorHandler::RecoverFromRetryableBGIOError() {
    EventHelpers::NotifyOnErrorRecoveryEnd(db_options_.listeners, bg_error_,
                                           Status::ShutdownInProgress(),
                                           db_mutex_);
+
+    recovery_in_prog_ = false;
    return;
  }
  DBRecoverContext context = recover_context_;
+  context.flush_after_recovery = true;
  int resume_count = db_options_.max_bgerror_resume_count;
  uint64_t wait_interval = db_options_.bgerror_resume_retry_interval;
  uint64_t retry_count = 0;
@@ -688,12 +704,12 @@ void ErrorHandler::RecoverFromRetryableBGIOError() {
      EventHelpers::NotifyOnErrorRecoveryEnd(db_options_.listeners, bg_error_,
                                             Status::ShutdownInProgress(),
                                             db_mutex_);
+      recovery_in_prog_ = false;
      return;
    }
    TEST_SYNC_POINT("RecoverFromRetryableBGIOError:BeforeResume0");
    TEST_SYNC_POINT("RecoverFromRetryableBGIOError:BeforeResume1");
-    recovery_io_error_ = IOStatus::OK();
-    recovery_error_ = Status::OK();
+    recovery_error_ = IOStatus::OK();
    retry_count++;
    Status s = db_->ResumeImpl(context);
    if (bg_error_stats_ != nullptr) {
@@ -713,9 +729,9 @@ void ErrorHandler::RecoverFromRetryableBGIOError() {
                                             bg_error_, db_mutex_);
      return;
    }
-    if (!recovery_io_error_.ok() &&
+    if (!recovery_error_.ok() &&
        recovery_error_.severity() <= Status::Severity::kHardError &&
-        recovery_io_error_.GetRetryable()) {
+        recovery_error_.GetRetryable()) {
      // If new BG IO error happens during auto recovery and it is retryable
      // and its severity is Hard Error or lower, the auto resmue sleep for
      // a period of time and redo auto resume if it is allowed.
@@ -724,10 +740,10 @@ void ErrorHandler::RecoverFromRetryableBGIOError() {
      int64_t wait_until = db_options_.clock->NowMicros() + wait_interval;
      cv_.TimedWait(wait_until);
    } else {
-      // There are three possibility: 1) recover_io_error is set during resume
+      // There are three possibility: 1) recovery_error_ is set during resume
      // and the error is not retryable, 2) recover is successful, 3) other
      // error happens during resume and cannot be resumed here.
-      if (recovery_io_error_.ok() && recovery_error_.ok() && s.ok()) {
+      if (recovery_error_.ok() && s.ok()) {
        // recover from the retryable IO error and no other BG errors. Clean
        // the bg_error and notify user.
        TEST_SYNC_POINT("RecoverFromRetryableBGIOError:RecoverSuccess");
@@ -749,8 +765,8 @@ void ErrorHandler::RecoverFromRetryableBGIOError() {
        }
        return;
      } else {
-        // In this case: 1) recovery_io_error is more serious or not retryable
-        // 2) other Non IO recovery_error happens. The auto recovery stops.
+        // In this case: 1) recovery_error_ is more serious or not retryable
+        // 2) other error happens. The auto recovery stops.
        recovery_in_prog_ = false;
        if (bg_error_stats_ != nullptr) {
          RecordInHistogram(bg_error_stats_.get(),
@@ -758,10 +774,7 @@ void ErrorHandler::RecoverFromRetryableBGIOError() {
        }
        EventHelpers::NotifyOnErrorRecoveryEnd(
            db_options_.listeners, bg_error_,
-            !recovery_io_error_.ok()
-                ? recovery_io_error_
-                : (!recovery_error_.ok() ? recovery_error_ : s),
-            db_mutex_);
+            !recovery_error_.ok() ? recovery_error_ : s, db_mutex_);
        return;
      }
    }
@@ -781,7 +794,7 @@ void ErrorHandler::RecoverFromRetryableBGIOError() {

 void ErrorHandler::CheckAndSetRecoveryAndBGError(const Status& bg_err) {
  if (recovery_in_prog_ && recovery_error_.ok()) {
-    recovery_error_ = bg_err;
+    recovery_error_ = status_to_io_status(Status(bg_err));
  }
  if (bg_err.severity() > bg_error_.severity()) {
    bg_error_ = bg_err;
@@ -797,12 +810,15 @@ void ErrorHandler::EndAutoRecovery() {
  if (!end_recovery_) {
    end_recovery_ = true;
  }
-  cv_.SignalAll();
-  db_mutex_->Unlock();
  if (recovery_thread_) {
-    recovery_thread_->join();
+    // Ensure only one thread can execute the join().
+    std::unique_ptr<port::Thread> old_recovery_thread(
+        std::move(recovery_thread_));
+    db_mutex_->Unlock();
+    cv_.SignalAll();
+    old_recovery_thread->join();
+    db_mutex_->Lock();
  }
-  db_mutex_->Lock();
  return;
 }

@@ -19,10 +19,13 @@ class DBImpl;
 // FlushReason, which tells the flush job why this flush is called.
 struct DBRecoverContext {
  FlushReason flush_reason;
+  bool flush_after_recovery;

-  DBRecoverContext() : flush_reason(FlushReason::kErrorRecovery) {}
-
-  DBRecoverContext(FlushReason reason) : flush_reason(reason) {}
+  DBRecoverContext()
+      : flush_reason(FlushReason::kErrorRecovery),
+        flush_after_recovery(false) {}
+  DBRecoverContext(FlushReason reason)
+      : flush_reason(reason), flush_after_recovery(false) {}
 };

 class ErrorHandler {
@@ -39,11 +42,11 @@ class ErrorHandler {
        recovery_in_prog_(false),
        soft_error_no_bg_work_(false),
        is_db_stopped_(false),
-        bg_error_stats_(db_options.statistics) {
+        bg_error_stats_(db_options.statistics),
+        recovery_disabled_file_deletion_(false) {
    // Clear the checked flag for uninitialized errors
    bg_error_.PermitUncheckedError();
    recovery_error_.PermitUncheckedError();
-    recovery_io_error_.PermitUncheckedError();
  }

  void EnableAutoRecovery() { auto_recovery_ = true; }
@@ -84,10 +87,7 @@ class ErrorHandler {
  Status bg_error_;
  // A separate Status variable used to record any errors during the
  // recovery process from hard errors
-  Status recovery_error_;
-  // A separate IO Status variable used to record any IO errors during
-  // the recovery process. At the same time, recovery_error_ is also set.
-  IOStatus recovery_io_error_;
+  IOStatus recovery_error_;
  // The condition variable used with db_mutex during auto resume for time
  // wait.
  InstrumentedCondVar cv_;
@@ -109,6 +109,10 @@ class ErrorHandler {
  // The pointer of DB statistics.
  std::shared_ptr<Statistics> bg_error_stats_;

+  // Tracks whether the recovery has disabled file deletion. This boolean flag
+  // is updated while holding db mutex.
+  bool recovery_disabled_file_deletion_;
+
  const Status& HandleKnownErrors(const Status& bg_err,
                                  BackgroundErrorReason reason);
  Status OverrideNoSpaceError(const Status& bg_error, bool* auto_recovery);
@@ -219,10 +219,16 @@ struct FileMetaData {
  // refers to. 0 is an invalid value; BlobDB numbers the files starting from 1.
  uint64_t oldest_blob_file_number = kInvalidBlobFileNumber;

-  // The file could be the compaction output from other SST files, which could
-  // in turn be outputs for compact older SST files. We track the memtable
-  // flush timestamp for the oldest SST file that eventually contribute data
-  // to this file. 0 means the information is not available.
+  // For flush output file, oldest ancestor time is the oldest key time in the
+  // file.  If the oldest key time is not available, flush time is used.
+  //
+  // For compaction output file, oldest ancestor time is the oldest
+  // among all the oldest key time of its input files, since the file could be
+  // the compaction output from other SST files, which could in turn be outputs
+  // for compact older SST files. If that's not available, creation time of this
+  // compaction output file is used.
+  //
+  // 0 means the information is not available.
  uint64_t oldest_ancester_time = kUnknownOldestAncesterTime;

  // Unix time when the SST file is created.
@@ -3582,26 +3582,16 @@ void VersionStorageInfo::ComputeCompactionScore(
    }
  }
  ComputeFilesMarkedForCompaction(max_output_level);
-  if (!immutable_options.allow_ingest_behind) {
-    ComputeBottommostFilesMarkedForCompaction();
-  }
-  if (mutable_cf_options.ttl > 0 &&
-      compaction_style_ == kCompactionStyleLevel) {
-    ComputeExpiredTtlFiles(immutable_options, mutable_cf_options.ttl);
-  }
-  if (mutable_cf_options.periodic_compaction_seconds > 0) {
-    ComputeFilesMarkedForPeriodicCompaction(
-        immutable_options, mutable_cf_options.periodic_compaction_seconds,
-        max_output_level);
-  }
-
-  if (mutable_cf_options.enable_blob_garbage_collection &&
-      mutable_cf_options.blob_garbage_collection_age_cutoff > 0.0 &&
-      mutable_cf_options.blob_garbage_collection_force_threshold < 1.0) {
-    ComputeFilesMarkedForForcedBlobGC(
-        mutable_cf_options.blob_garbage_collection_age_cutoff,
-        mutable_cf_options.blob_garbage_collection_force_threshold);
-  }
+  ComputeBottommostFilesMarkedForCompaction(
+      immutable_options.allow_ingest_behind);
+  ComputeExpiredTtlFiles(immutable_options, mutable_cf_options.ttl);
+  ComputeFilesMarkedForPeriodicCompaction(
+      immutable_options, mutable_cf_options.periodic_compaction_seconds,
+      max_output_level);
+  ComputeFilesMarkedForForcedBlobGC(
+      mutable_cf_options.blob_garbage_collection_age_cutoff,
+      mutable_cf_options.blob_garbage_collection_force_threshold,
+      mutable_cf_options.enable_blob_garbage_collection);

  EstimateCompactionBytesNeeded(mutable_cf_options);
 }
@@ -3631,9 +3621,10 @@ void VersionStorageInfo::ComputeFilesMarkedForCompaction(int last_level) {

 void VersionStorageInfo::ComputeExpiredTtlFiles(
    const ImmutableOptions& ioptions, const uint64_t ttl) {
-  assert(ttl > 0);
-
  expired_ttl_files_.clear();
+  if (ttl == 0 || compaction_style_ != CompactionStyle::kCompactionStyleLevel) {
+    return;
+  }

  int64_t _current_time;
  auto status = ioptions.clock->GetCurrentTime(&_current_time);
@@ -3658,9 +3649,10 @@ void VersionStorageInfo::ComputeExpiredTtlFiles(
 void VersionStorageInfo::ComputeFilesMarkedForPeriodicCompaction(
    const ImmutableOptions& ioptions,
    const uint64_t periodic_compaction_seconds, int last_level) {
-  assert(periodic_compaction_seconds > 0);
-
  files_marked_for_periodic_compaction_.clear();
+  if (periodic_compaction_seconds == 0) {
+    return;
+  }

  int64_t temp_current_time;
  auto status = ioptions.clock->GetCurrentTime(&temp_current_time);
@@ -3714,8 +3706,14 @@ void VersionStorageInfo::ComputeFilesMarkedForPeriodicCompaction(

 void VersionStorageInfo::ComputeFilesMarkedForForcedBlobGC(
    double blob_garbage_collection_age_cutoff,
-    double blob_garbage_collection_force_threshold) {
+    double blob_garbage_collection_force_threshold,
+    bool enable_blob_garbage_collection) {
  files_marked_for_forced_blob_gc_.clear();
+  if (!(enable_blob_garbage_collection &&
+        blob_garbage_collection_age_cutoff > 0.0 &&
+        blob_garbage_collection_force_threshold < 1.0)) {
+    return;
+  }

  if (blob_files_.empty()) {
    return;
@@ -4172,17 +4170,22 @@ void VersionStorageInfo::GenerateFileLocationIndex() {
  }
 }

-void VersionStorageInfo::UpdateOldestSnapshot(SequenceNumber seqnum) {
+void VersionStorageInfo::UpdateOldestSnapshot(SequenceNumber seqnum,
+                                              bool allow_ingest_behind) {
  assert(seqnum >= oldest_snapshot_seqnum_);
  oldest_snapshot_seqnum_ = seqnum;
  if (oldest_snapshot_seqnum_ > bottommost_files_mark_threshold_) {
-    ComputeBottommostFilesMarkedForCompaction();
+    ComputeBottommostFilesMarkedForCompaction(allow_ingest_behind);
  }
 }

-void VersionStorageInfo::ComputeBottommostFilesMarkedForCompaction() {
+void VersionStorageInfo::ComputeBottommostFilesMarkedForCompaction(
+    bool allow_ingest_behind) {
  bottommost_files_marked_for_compaction_.clear();
  bottommost_files_mark_threshold_ = kMaxSequenceNumber;
+  if (allow_ingest_behind) {
+    return;
+  }
  // If a file's creation time is larger than creation_time_ub,
  // it is too new to be marked for compaction.
  int64_t creation_time_ub = 0;
@@ -228,7 +228,7 @@ class VersionStorageInfo {
  // eligible for compaction.
  //
  // REQUIRES: DB mutex held
-  void ComputeBottommostFilesMarkedForCompaction();
+  void ComputeBottommostFilesMarkedForCompaction(bool allow_ingest_behind);

  // This computes files_marked_for_forced_blob_gc_ and is called by
  // ComputeCompactionScore()
@@ -236,14 +236,16 @@ class VersionStorageInfo {
  // REQUIRES: DB mutex held
  void ComputeFilesMarkedForForcedBlobGC(
      double blob_garbage_collection_age_cutoff,
-      double blob_garbage_collection_force_threshold);
+      double blob_garbage_collection_force_threshold,
+      bool enable_blob_garbage_collection);

  bool level0_non_overlapping() const { return level0_non_overlapping_; }

  // Updates the oldest snapshot and related internal state, like the bottommost
  // files marked for compaction.
  // REQUIRES: DB mutex held
-  void UpdateOldestSnapshot(SequenceNumber oldest_snapshot_seqnum);
+  void UpdateOldestSnapshot(SequenceNumber oldest_snapshot_seqnum,
+                            bool allow_ingest_behind);

  int MaxInputLevel() const;
  int MaxOutputLevel(bool allow_ingest_behind) const;
@@ -584,7 +584,8 @@ TEST_F(VersionStorageInfoTest, ForcedBlobGCEmpty) {

  constexpr double age_cutoff = 0.5;
  constexpr double force_threshold = 0.75;
-  vstorage_.ComputeFilesMarkedForForcedBlobGC(age_cutoff, force_threshold);
+  vstorage_.ComputeFilesMarkedForForcedBlobGC(
+      age_cutoff, force_threshold, /*enable_blob_garbage_collection=*/true);

  ASSERT_TRUE(vstorage_.FilesMarkedForForcedBlobGC().empty());
 }
@@ -668,7 +669,8 @@ TEST_F(VersionStorageInfoTest, ForcedBlobGCSingleBatch) {
  {
    constexpr double age_cutoff = 0.1;
    constexpr double force_threshold = 0.0;
-    vstorage_.ComputeFilesMarkedForForcedBlobGC(age_cutoff, force_threshold);
+    vstorage_.ComputeFilesMarkedForForcedBlobGC(
+        age_cutoff, force_threshold, /*enable_blob_garbage_collection=*/true);

    ASSERT_TRUE(vstorage_.FilesMarkedForForcedBlobGC().empty());
  }
@@ -679,7 +681,8 @@ TEST_F(VersionStorageInfoTest, ForcedBlobGCSingleBatch) {
  {
    constexpr double age_cutoff = 0.5;
    constexpr double force_threshold = 0.0;
-    vstorage_.ComputeFilesMarkedForForcedBlobGC(age_cutoff, force_threshold);
+    vstorage_.ComputeFilesMarkedForForcedBlobGC(
+        age_cutoff, force_threshold, /*enable_blob_garbage_collection=*/true);

    ASSERT_TRUE(vstorage_.FilesMarkedForForcedBlobGC().empty());
  }
@@ -690,7 +693,8 @@ TEST_F(VersionStorageInfoTest, ForcedBlobGCSingleBatch) {
  {
    constexpr double age_cutoff = 1.0;
    constexpr double force_threshold = 0.6;
-    vstorage_.ComputeFilesMarkedForForcedBlobGC(age_cutoff, force_threshold);
+    vstorage_.ComputeFilesMarkedForForcedBlobGC(
+        age_cutoff, force_threshold, /*enable_blob_garbage_collection=*/true);

    ASSERT_TRUE(vstorage_.FilesMarkedForForcedBlobGC().empty());
  }
@@ -701,7 +705,8 @@ TEST_F(VersionStorageInfoTest, ForcedBlobGCSingleBatch) {
  {
    constexpr double age_cutoff = 1.0;
    constexpr double force_threshold = 0.5;
-    vstorage_.ComputeFilesMarkedForForcedBlobGC(age_cutoff, force_threshold);
+    vstorage_.ComputeFilesMarkedForForcedBlobGC(
+        age_cutoff, force_threshold, /*enable_blob_garbage_collection=*/true);

    auto ssts_to_be_compacted = vstorage_.FilesMarkedForForcedBlobGC();
    ASSERT_EQ(ssts_to_be_compacted.size(), 1);
@@ -815,7 +820,8 @@ TEST_F(VersionStorageInfoTest, ForcedBlobGCMultipleBatches) {
  {
    constexpr double age_cutoff = 0.1;
    constexpr double force_threshold = 0.0;
-    vstorage_.ComputeFilesMarkedForForcedBlobGC(age_cutoff, force_threshold);
+    vstorage_.ComputeFilesMarkedForForcedBlobGC(
+        age_cutoff, force_threshold, /*enable_blob_garbage_collection=*/true);

    ASSERT_TRUE(vstorage_.FilesMarkedForForcedBlobGC().empty());
  }
@@ -826,7 +832,8 @@ TEST_F(VersionStorageInfoTest, ForcedBlobGCMultipleBatches) {
  {
    constexpr double age_cutoff = 0.25;
    constexpr double force_threshold = 0.0;
-    vstorage_.ComputeFilesMarkedForForcedBlobGC(age_cutoff, force_threshold);
+    vstorage_.ComputeFilesMarkedForForcedBlobGC(
+        age_cutoff, force_threshold, /*enable_blob_garbage_collection=*/true);

    ASSERT_TRUE(vstorage_.FilesMarkedForForcedBlobGC().empty());
  }
@@ -837,7 +844,8 @@ TEST_F(VersionStorageInfoTest, ForcedBlobGCMultipleBatches) {
  {
    constexpr double age_cutoff = 0.5;
    constexpr double force_threshold = 0.6;
-    vstorage_.ComputeFilesMarkedForForcedBlobGC(age_cutoff, force_threshold);
+    vstorage_.ComputeFilesMarkedForForcedBlobGC(
+        age_cutoff, force_threshold, /*enable_blob_garbage_collection=*/true);

    ASSERT_TRUE(vstorage_.FilesMarkedForForcedBlobGC().empty());
  }
@@ -848,7 +856,8 @@ TEST_F(VersionStorageInfoTest, ForcedBlobGCMultipleBatches) {
  {
    constexpr double age_cutoff = 0.5;
    constexpr double force_threshold = 0.5;
-    vstorage_.ComputeFilesMarkedForForcedBlobGC(age_cutoff, force_threshold);
+    vstorage_.ComputeFilesMarkedForForcedBlobGC(
+        age_cutoff, force_threshold, /*enable_blob_garbage_collection=*/true);

    auto ssts_to_be_compacted = vstorage_.FilesMarkedForForcedBlobGC();
    ASSERT_EQ(ssts_to_be_compacted.size(), 2);
@@ -877,7 +886,8 @@ TEST_F(VersionStorageInfoTest, ForcedBlobGCMultipleBatches) {
  {
    constexpr double age_cutoff = 0.75;
    constexpr double force_threshold = 0.6;
-    vstorage_.ComputeFilesMarkedForForcedBlobGC(age_cutoff, force_threshold);
+    vstorage_.ComputeFilesMarkedForForcedBlobGC(
+        age_cutoff, force_threshold, /*enable_blob_garbage_collection=*/true);

    ASSERT_TRUE(vstorage_.FilesMarkedForForcedBlobGC().empty());
  }
@@ -888,7 +898,8 @@ TEST_F(VersionStorageInfoTest, ForcedBlobGCMultipleBatches) {
  {
    constexpr double age_cutoff = 0.75;
    constexpr double force_threshold = 0.5;
-    vstorage_.ComputeFilesMarkedForForcedBlobGC(age_cutoff, force_threshold);
+    vstorage_.ComputeFilesMarkedForForcedBlobGC(
+        age_cutoff, force_threshold, /*enable_blob_garbage_collection=*/true);

    auto ssts_to_be_compacted = vstorage_.FilesMarkedForForcedBlobGC();
    ASSERT_EQ(ssts_to_be_compacted.size(), 2);
@@ -13,6 +13,7 @@

 #include <cmath>

+#include "rocksdb/secondary_cache.h"
 #include "util/file_checksum_helper.h"
 #include "util/xxhash.h"

@@ -21,6 +22,8 @@ ROCKSDB_NAMESPACE::Env* db_stress_env = nullptr;
 // If non-null, injects read error at a rate specified by the
 // read_fault_one_in or write_fault_one_in flag
 std::shared_ptr<ROCKSDB_NAMESPACE::FaultInjectionTestFS> fault_fs_guard;
+std::shared_ptr<ROCKSDB_NAMESPACE::SecondaryCache> compressed_secondary_cache;
+std::shared_ptr<ROCKSDB_NAMESPACE::Cache> block_cache;
 enum ROCKSDB_NAMESPACE::CompressionType compression_type_e =
    ROCKSDB_NAMESPACE::kSnappyCompression;
 enum ROCKSDB_NAMESPACE::CompressionType bottommost_compression_type_e =
@@ -148,6 +151,88 @@ void DbVerificationThread(void* v) {
  }
 }

+void CompressedCacheSetCapacityThread(void* v) {
+  assert(FLAGS_compressed_secondary_cache_size > 0 ||
+         FLAGS_compressed_secondary_cache_ratio > 0.0);
+  auto* thread = reinterpret_cast<ThreadState*>(v);
+  SharedState* shared = thread->shared;
+  while (true) {
+    {
+      MutexLock l(shared->GetMutex());
+      if (shared->ShouldStopBgThread()) {
+        shared->IncBgThreadsFinished();
+        if (shared->BgThreadsFinished()) {
+          shared->GetCondVar()->SignalAll();
+        }
+        return;
+      }
+    }
+    db_stress_env->SleepForMicroseconds(FLAGS_secondary_cache_update_interval);
+    if (FLAGS_compressed_secondary_cache_size > 0) {
+      Status s = compressed_secondary_cache->SetCapacity(0);
+      size_t capacity;
+      if (s.ok()) {
+        s = compressed_secondary_cache->GetCapacity(capacity);
+        assert(capacity == 0);
+      }
+      db_stress_env->SleepForMicroseconds(10 * 1000 * 1000);
+      if (s.ok()) {
+        s = compressed_secondary_cache->SetCapacity(
+            FLAGS_compressed_secondary_cache_size);
+      }
+      if (s.ok()) {
+        s = compressed_secondary_cache->GetCapacity(capacity);
+        assert(capacity == FLAGS_compressed_secondary_cache_size);
+      }
+      if (!s.ok()) {
+        fprintf(stderr, "Compressed cache Set/GetCapacity returned error: %s\n",
+                s.ToString().c_str());
+      }
+    } else if (FLAGS_compressed_secondary_cache_ratio > 0.0) {
+      if (thread->rand.OneIn(2)) {
+        size_t capacity = block_cache->GetCapacity();
+        size_t adjustment;
+        if (FLAGS_use_write_buffer_manager && FLAGS_db_write_buffer_size > 0) {
+          adjustment = (capacity - FLAGS_db_write_buffer_size);
+        } else {
+          adjustment = capacity;
+        }
+        // Lower by upto 50% of usable block cache capacity
+        adjustment = (adjustment * thread->rand.Uniform(50)) / 100;
+        block_cache->SetCapacity(capacity - adjustment);
+        fprintf(stderr, "New cache capacity = %lu\n",
+                block_cache->GetCapacity());
+        db_stress_env->SleepForMicroseconds(10 * 1000 * 1000);
+        block_cache->SetCapacity(capacity);
+      } else {
+        Status s;
+        double new_comp_cache_ratio =
+            (double)thread->rand.Uniform(
+                FLAGS_compressed_secondary_cache_ratio * 100) /
+            100;
+        if (new_comp_cache_ratio == 0.0) {
+          new_comp_cache_ratio = 0.05;
+        }
+        fprintf(stderr, "New comp cache ratio = %f\n", new_comp_cache_ratio);
+
+        s = UpdateTieredCache(block_cache, /*capacity*/ -1,
+                              new_comp_cache_ratio);
+        if (s.ok()) {
+          db_stress_env->SleepForMicroseconds(10 * 1000 * 1000);
+        }
+        if (s.ok()) {
+          s = UpdateTieredCache(block_cache, /*capacity*/ -1,
+                                FLAGS_compressed_secondary_cache_ratio);
+        }
+        if (!s.ok()) {
+          fprintf(stderr, "UpdateTieredCache returned error: %s\n",
+                  s.ToString().c_str());
+        }
+      }
+    }
+  }
+}
+
 void PrintKeyValue(int cf, uint64_t key, const char* value, size_t sz) {
  if (!FLAGS_verbose) {
    return;
@@ -108,11 +108,14 @@ DECLARE_int32(max_write_buffer_number);
 DECLARE_int32(min_write_buffer_number_to_merge);
 DECLARE_int32(max_write_buffer_number_to_maintain);
 DECLARE_int64(max_write_buffer_size_to_maintain);
+DECLARE_bool(use_write_buffer_manager);
 DECLARE_double(memtable_prefix_bloom_size_ratio);
 DECLARE_bool(memtable_whole_key_filtering);
 DECLARE_int32(open_files);
-DECLARE_int64(compressed_cache_size);
-DECLARE_int32(compressed_cache_numshardbits);
+DECLARE_uint64(compressed_secondary_cache_size);
+DECLARE_int32(compressed_secondary_cache_numshardbits);
+DECLARE_int32(secondary_cache_update_interval);
+DECLARE_double(compressed_secondary_cache_ratio);
 DECLARE_int32(compaction_style);
 DECLARE_int32(compaction_pri);
 DECLARE_int32(num_levels);
@@ -358,6 +361,9 @@ constexpr int kValueMaxLen = 100;
 extern ROCKSDB_NAMESPACE::Env* db_stress_env;
 extern ROCKSDB_NAMESPACE::Env* db_stress_listener_env;
 extern std::shared_ptr<ROCKSDB_NAMESPACE::FaultInjectionTestFS> fault_fs_guard;
+extern std::shared_ptr<ROCKSDB_NAMESPACE::SecondaryCache>
+    compressed_secondary_cache;
+extern std::shared_ptr<ROCKSDB_NAMESPACE::Cache> block_cache;

 extern enum ROCKSDB_NAMESPACE::CompressionType compression_type_e;
 extern enum ROCKSDB_NAMESPACE::CompressionType bottommost_compression_type_e;
@@ -650,6 +656,8 @@ extern void PoolSizeChangeThread(void* v);

 extern void DbVerificationThread(void* v);

+extern void CompressedCacheSetCapacityThread(void* v);
+
 extern void TimestampedSnapshotsThread(void* v);

 extern void PrintKeyValue(int cf, uint64_t key, const char* value, size_t sz);
@@ -121,6 +121,11 @@ bool RunStressTestImpl(SharedState* shared) {
    shared->IncBgThreads();
  }

+  if (FLAGS_compressed_secondary_cache_size > 0 ||
+      FLAGS_compressed_secondary_cache_ratio > 0.0) {
+    shared->IncBgThreads();
+  }
+
  std::vector<ThreadState*> threads(n);
  for (uint32_t i = 0; i < n; i++) {
    threads[i] = new ThreadState(i, shared);
@@ -138,6 +143,13 @@ bool RunStressTestImpl(SharedState* shared) {
                               &continuous_verification_thread);
  }

+  ThreadState compressed_cache_set_capacity_thread(0, shared);
+  if (FLAGS_compressed_secondary_cache_size > 0 ||
+      FLAGS_compressed_secondary_cache_ratio > 0.0) {
+    db_stress_env->StartThread(CompressedCacheSetCapacityThread,
+                               &compressed_cache_set_capacity_thread);
+  }
+
  // Each thread goes through the following states:
  // initializing -> wait for others to init -> read/populate/depopulate
  // wait for others to operate -> verify -> done
@@ -230,7 +242,9 @@ bool RunStressTestImpl(SharedState* shared) {
  }

  if (FLAGS_compaction_thread_pool_adjust_interval > 0 ||
-      FLAGS_continuous_verification_interval > 0) {
+      FLAGS_continuous_verification_interval > 0 ||
+      FLAGS_compressed_secondary_cache_size > 0 ||
+      FLAGS_compressed_secondary_cache_ratio > 0.0) {
    MutexLock l(shared->GetMutex());
    shared->SetShouldStopBgThread();
    while (!shared->BgThreadsFinished()) {
@@ -136,6 +136,9 @@ DEFINE_uint64(db_write_buffer_size,
              ROCKSDB_NAMESPACE::Options().db_write_buffer_size,
              "Number of bytes to buffer in all memtables before compacting");

+DEFINE_bool(use_write_buffer_manager, false,
+            "Charge WriteBufferManager memory to the block cache");
+
 DEFINE_int32(
    write_buffer_size,
    static_cast<int32_t>(ROCKSDB_NAMESPACE::Options().write_buffer_size),
@@ -198,15 +201,23 @@ DEFINE_int32(open_files, ROCKSDB_NAMESPACE::Options().max_open_files,
             "Maximum number of files to keep open at the same time "
             "(use default if == 0)");

-DEFINE_int64(compressed_cache_size, 0,
-             "Number of bytes to use as a cache of compressed data."
-             " 0 means use default settings.");
+DEFINE_uint64(compressed_secondary_cache_size, 0,
+              "Number of bytes to use as a cache of compressed data."
+              " 0 means use default settings.");

-DEFINE_int32(
-    compressed_cache_numshardbits, -1,
-    "Number of shards for the compressed block cache is 2 ** "
-    "compressed_cache_numshardbits. Negative value means default settings. "
-    "This is applied only if compressed_cache_size is greater than 0.");
+DEFINE_int32(compressed_secondary_cache_numshardbits, -1,
+             "Number of shards for the compressed secondary cache is 2 ** "
+             "compressed_secondary_cache_numshardbits. "
+             "Negative value means default settings. This is applied only "
+             "if compressed_secondary_cache_size is greater than 0.");
+
+DEFINE_double(compressed_secondary_cache_ratio, 0.0,
+              "Fraction of block cache memory budget to use for compressed "
+              "secondary cache");
+
+DEFINE_int32(secondary_cache_update_interval, 30 * 1000 * 1000,
+             "Interval between modification of secondary cache parameters, in "
+             "microseconds");

 DEFINE_int32(compaction_style, ROCKSDB_NAMESPACE::Options().compaction_style,
             "");
@@ -1023,6 +1034,9 @@ DEFINE_string(secondary_cache_uri, "",
 DEFINE_int32(secondary_cache_fault_one_in, 0,
             "On non-zero, enables fault injection in secondary cache inserts"
             " and lookups");
+DEFINE_double(tiered_cache_percent_compressed, 0.0,
+              "Percentage of total block cache budget to allocate to the "
+              "compressed cache");
 DEFINE_int32(open_write_fault_one_in, 0,
             "On non-zero, enables fault injection on file writes "
             "during DB reopen.");
@@ -111,6 +111,11 @@ std::shared_ptr<Cache> StressTest::NewCache(size_t capacity,

  std::shared_ptr<SecondaryCache> secondary_cache;
  if (!FLAGS_secondary_cache_uri.empty()) {
+    assert(!strstr(FLAGS_secondary_cache_uri.c_str(),
+                   "compressed_secondary_cache") ||
+           (FLAGS_compressed_secondary_cache_size == 0 &&
+            FLAGS_compressed_secondary_cache_ratio == 0.0 &&
+            !StartsWith(FLAGS_cache_type, "tiered_")));
    Status s = SecondaryCache::CreateFromString(
        config_options, FLAGS_secondary_cache_uri, &secondary_cache);
    if (secondary_cache == nullptr) {
@@ -124,36 +129,81 @@ std::shared_ptr<Cache> StressTest::NewCache(size_t capacity,
          secondary_cache, static_cast<uint32_t>(FLAGS_seed),
          FLAGS_secondary_cache_fault_one_in);
    }
+  } else if (FLAGS_compressed_secondary_cache_size > 0) {
+    if (StartsWith(FLAGS_cache_type, "tiered_")) {
+      fprintf(stderr,
+              "Cannot specify both compressed_secondary_cache_size and %s\n",
+              FLAGS_cache_type.c_str());
+      exit(1);
+    }
+    CompressedSecondaryCacheOptions opts;
+    opts.capacity = FLAGS_compressed_secondary_cache_size;
+    secondary_cache = NewCompressedSecondaryCache(opts);
+    if (secondary_cache == nullptr) {
+      fprintf(stderr, "Failed to allocate compressed secondary cache\n");
+      exit(1);
+    }
+    compressed_secondary_cache = secondary_cache;
  }

-  if (FLAGS_cache_type == "clock_cache") {
+  std::string cache_type = FLAGS_cache_type;
+  size_t cache_size = FLAGS_cache_size;
+  bool tiered = false;
+  if (StartsWith(cache_type, "tiered_")) {
+    tiered = true;
+    cache_type.erase(0, strlen("tiered_"));
+  }
+  if (FLAGS_use_write_buffer_manager) {
+    cache_size += FLAGS_db_write_buffer_size;
+  }
+  if (cache_type == "clock_cache") {
    fprintf(stderr, "Old clock cache implementation has been removed.\n");
    exit(1);
-  } else if (EndsWith(FLAGS_cache_type, "hyper_clock_cache")) {
+  } else if (EndsWith(cache_type, "hyper_clock_cache")) {
    size_t estimated_entry_charge;
-    if (FLAGS_cache_type == "fixed_hyper_clock_cache" ||
-        FLAGS_cache_type == "hyper_clock_cache") {
+    if (cache_type == "fixed_hyper_clock_cache" ||
+        cache_type == "hyper_clock_cache") {
      estimated_entry_charge = FLAGS_block_size;
-    } else if (FLAGS_cache_type == "auto_hyper_clock_cache") {
+    } else if (cache_type == "auto_hyper_clock_cache") {
      estimated_entry_charge = 0;
    } else {
      fprintf(stderr, "Cache type not supported.");
      exit(1);
    }
-    HyperClockCacheOptions opts(FLAGS_cache_size, estimated_entry_charge,
+    HyperClockCacheOptions opts(cache_size, estimated_entry_charge,
                                num_shard_bits);
    opts.hash_seed = BitwiseAnd(FLAGS_seed, INT32_MAX);
-    return opts.MakeSharedCache();
-  } else if (FLAGS_cache_type == "lru_cache") {
+    if (tiered) {
+      TieredCacheOptions tiered_opts;
+      tiered_opts.cache_opts = &opts;
+      tiered_opts.cache_type = PrimaryCacheType::kCacheTypeHCC;
+      tiered_opts.total_capacity = cache_size;
+      tiered_opts.compressed_secondary_ratio = 0.5;
+      block_cache = NewTieredCache(tiered_opts);
+    } else {
+      opts.secondary_cache = std::move(secondary_cache);
+      block_cache = opts.MakeSharedCache();
+    }
+  } else if (EndsWith(cache_type, "lru_cache")) {
    LRUCacheOptions opts;
    opts.capacity = capacity;
    opts.num_shard_bits = num_shard_bits;
-    opts.secondary_cache = std::move(secondary_cache);
-    return NewLRUCache(opts);
+    if (tiered) {
+      TieredCacheOptions tiered_opts;
+      tiered_opts.cache_opts = &opts;
+      tiered_opts.cache_type = PrimaryCacheType::kCacheTypeLRU;
+      tiered_opts.total_capacity = cache_size;
+      tiered_opts.compressed_secondary_ratio = 0.5;
+      block_cache = NewTieredCache(tiered_opts);
+    } else {
+      opts.secondary_cache = std::move(secondary_cache);
+      block_cache = NewLRUCache(opts);
+    }
  } else {
    fprintf(stderr, "Cache type not supported.");
    exit(1);
  }
+  return block_cache;
 }

 std::vector<std::string> StressTest::GetBlobCompressionTags() {
@@ -1235,7 +1285,6 @@ Status StressTest::TestIterate(ThreadState* thread,
  } else if (options_.prefix_extractor.get() == nullptr) {
    expect_total_order = true;
  }
-
  std::string upper_bound_str;
  Slice upper_bound;
  if (thread->rand.OneIn(16)) {
@@ -1246,6 +1295,7 @@ Status StressTest::TestIterate(ThreadState* thread,
    upper_bound = Slice(upper_bound_str);
    ro.iterate_upper_bound = &upper_bound;
  }
+
  std::string lower_bound_str;
  Slice lower_bound;
  if (thread->rand.OneIn(16)) {
@@ -1375,8 +1425,7 @@ Status StressTest::TestIterate(ThreadState* thread,
                   key, op_logs, &diverged);

    const bool no_reverse =
-        (FLAGS_memtablerep == "prefix_hash" && !expect_total_order) ||
-        FLAGS_auto_readahead_size;
+        (FLAGS_memtablerep == "prefix_hash" && !expect_total_order);
    for (uint64_t i = 0; i < FLAGS_num_iterations && iter->Valid(); ++i) {
      if (no_reverse || thread->rand.OneIn(2)) {
        iter->Next();
@@ -1564,7 +1613,8 @@ void StressTest::VerifyIterator(ThreadState* thread,
          fprintf(stderr, "iterator has value %s\n",
                  iter->key().ToString(true).c_str());
        } else {
-          fprintf(stderr, "iterator is not valid\n");
+          fprintf(stderr, "iterator is not valid with status: %s\n",
+                  iter->status().ToString().c_str());
        }
        *diverged = true;
      }
@@ -3174,6 +3224,10 @@ void InitializeOptionsFromFlags(
      FLAGS_max_write_buffer_size_to_maintain;
  options.memtable_prefix_bloom_size_ratio =
      FLAGS_memtable_prefix_bloom_size_ratio;
+  if (FLAGS_use_write_buffer_manager) {
+    options.write_buffer_manager.reset(
+        new WriteBufferManager(FLAGS_db_write_buffer_size, block_cache));
+  }
  options.memtable_whole_key_filtering = FLAGS_memtable_whole_key_filtering;
  options.disable_auto_compactions = FLAGS_disable_auto_compactions;
  options.max_background_compactions = FLAGS_max_background_compactions;
@@ -1914,8 +1914,7 @@ class NonBatchedOpsStressTest : public StressTest {
      if (static_cast<int64_t>(curr) < lb) {
        iter->Next();
        op_logs += "N";
-      } else if (static_cast<int64_t>(curr) >= ub &&
-                 !FLAGS_auto_readahead_size) {
+      } else if (static_cast<int64_t>(curr) >= ub) {
        iter->Prev();
        op_logs += "P";
      } else {
@@ -21,6 +21,7 @@
 #ifdef __SSE4_2__
 #ifdef _WIN32
 #include <intrin.h>
+#define _rdtsc() __rdtsc()
 #else
 #include <x86intrin.h>
 #endif
@@ -1448,21 +1448,49 @@ TEST_F(PrefetchTest, PrefetchWithBlockLookupAutoTuneWithPrev) {
    ropts.iterate_upper_bound = ub_ptr;
    ropts.auto_readahead_size = true;

+    ReadOptions cmp_readopts = ropts;
+    cmp_readopts.auto_readahead_size = false;
+
    auto iter = std::unique_ptr<Iterator>(db_->NewIterator(ropts));
+    auto cmp_iter = std::unique_ptr<Iterator>(db_->NewIterator(cmp_readopts));

    Slice seek_key = Slice("my_key_bbb");
-    iter->Seek(seek_key);
-    ASSERT_TRUE(iter->Valid());
+    {
+      cmp_iter->Seek(seek_key);
+      ASSERT_TRUE(cmp_iter->Valid());
+      ASSERT_OK(cmp_iter->status());

-    // Prev op should fail with auto tuning of readahead_size.
-    iter->Prev();
-    ASSERT_TRUE(iter->status().IsNotSupported());
-    ASSERT_FALSE(iter->Valid());
+      iter->Seek(seek_key);
+      ASSERT_TRUE(iter->Valid());
+      ASSERT_OK(iter->status());
+
+      ASSERT_EQ(iter->key(), cmp_iter->key());
+    }
+
+    // Prev op should pass with auto tuning of readahead_size.
+    {
+      cmp_iter->Prev();
+      ASSERT_TRUE(cmp_iter->Valid());
+      ASSERT_OK(cmp_iter->status());
+
+      iter->Prev();
+      ASSERT_OK(iter->status());
+      ASSERT_TRUE(iter->Valid());
+
+      ASSERT_EQ(iter->key(), cmp_iter->key());
+    }

    // Reseek would follow as usual.
-    iter->Seek(seek_key);
-    ASSERT_OK(iter->status());
-    ASSERT_TRUE(iter->Valid());
+    {
+      cmp_iter->Seek(seek_key);
+      ASSERT_TRUE(cmp_iter->Valid());
+      ASSERT_OK(cmp_iter->status());
+
+      iter->Seek(seek_key);
+      ASSERT_OK(iter->status());
+      ASSERT_TRUE(iter->Valid());
+      ASSERT_EQ(iter->key(), cmp_iter->key());
+    }
  }
  Close();
 }
@@ -375,6 +375,14 @@ class Cache {
  // Returns the helper for the specified entry.
  virtual const CacheItemHelper* GetCacheItemHelper(Handle* handle) const = 0;

+  virtual Status GetSecondaryCacheCapacity(size_t& /*size*/) const {
+    return Status::NotSupported();
+  }
+
+  virtual Status GetSecondaryCachePinnedUsage(size_t& /*size*/) const {
+    return Status::NotSupported();
+  }
+
  // Call this on shutdown if you want to speed it up. Cache will disown
  // any underlying data and will not free it on delete. This call will leak
  // memory - call this only if you're shutting down the process.
@@ -120,6 +120,7 @@ struct IOOptions {
  // directories and list only files in GetChildren API.
  bool do_not_recurse;

+  // EXPERIMENTAL
  Env::IOActivity io_activity = Env::IOActivity::kUnknown;

  IOOptions() : IOOptions(false) {}
@@ -1719,16 +1719,27 @@ struct ReadOptions {
  // during scans internally.
  // For this feature to enabled, iterate_upper_bound must also be specified.
  //
-  // NOTE: Not supported with Prev operation and it will be return NotSupported
-  // error. Enable it for forward scans only.
+  // NOTE: - Recommended for forward Scans only.
+  //       - In case of backward scans like Prev or SeekForPrev, the
+  //          cost of these backward operations might increase and affect the
+  //          performace. So this option should not be enabled if workload
+  //          contains backward scans.
+  //       - If there is a backward scans, this option will be
+  //          disabled internally and won't be reset if forward scan is done
+  //          again.
+  //
  // Default: false
  bool auto_readahead_size = false;

  // *** END options only relevant to iterators or scans ***

-  // ** For RocksDB internal use only **
+  // *** BEGIN options for RocksDB internal use only ***
+
+  // EXPERIMENTAL
  Env::IOActivity io_activity = Env::IOActivity::kUnknown;

+  // *** END options for RocksDB internal use only ***
+
  ReadOptions() {}
  ReadOptions(bool _verify_checksums, bool _fill_cache);
  explicit ReadOptions(Env::IOActivity _io_activity);
@@ -219,9 +219,20 @@ struct TableProperties {
  // by column_family_name.
  uint64_t column_family_id = ROCKSDB_NAMESPACE::
      TablePropertiesCollectorFactory::Context::kUnknownColumnFamily;
-  // Timestamp of the latest key. 0 means unknown.
-  // TODO(sagar0): Should be changed to latest_key_time ... but don't know the
-  // full implications of backward compatibility. Hence retaining for now.
+
+  // Oldest ancester time. 0 means unknown.
+  //
+  // For flush output file, oldest ancestor time is the oldest key time in the
+  // file.  If the oldest key time is not available, flush time is used.
+  //
+  // For compaction output file, oldest ancestor time is the oldest
+  // among all the oldest key time of its input files, since the file could be
+  // the compaction output from other SST files, which could in turn be outputs
+  // for compact older SST files. If that's not available, creation time of this
+  // compaction output file is used.
+  //
+  // TODO(sagar0): Should be changed to oldest_ancester_time ... but don't know
+  // the full implications of backward compatibility. Hence retaining for now.
  uint64_t creation_time = 0;

  // Timestamp of the earliest key. 0 means unknown.
@@ -13,7 +13,7 @@
 // minor or major version number planned for release.
 #define ROCKSDB_MAJOR 8
 #define ROCKSDB_MINOR 7
-#define ROCKSDB_PATCH 0
+#define ROCKSDB_PATCH 3

 // 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
@@ -26,7 +26,7 @@ void BlockBasedTableIterator::SeekImpl(const Slice* target,

  if (autotune_readaheadsize &&
      table_->get_rep()->table_options.block_cache.get() &&
-      !read_options_.async_io) {
+      !read_options_.async_io && direction_ == IterDirection::kForward) {
    readahead_cache_lookup_ = true;
  }

@@ -87,15 +87,13 @@ void BlockBasedTableIterator::SeekImpl(const Slice* target,
    } else {
      index_iter_->SeekToFirst();
    }
-
+    is_index_at_curr_block_ = true;
    if (!index_iter_->Valid()) {
      ResetDataIter();
      return;
    }
  }

-  is_index_at_curr_block_ = true;
-
  if (autotune_readaheadsize) {
    FindReadAheadSizeUpperBound();
    if (target) {
@@ -170,6 +168,8 @@ void BlockBasedTableIterator::SeekImpl(const Slice* target,
 }

 void BlockBasedTableIterator::SeekForPrev(const Slice& target) {
+  direction_ = IterDirection::kBackward;
+  ResetBlockCacheLookupVar();
  is_out_of_bound_ = false;
  is_at_first_key_from_index_ = false;
  seek_stat_state_ = kNone;
@@ -191,7 +191,6 @@ void BlockBasedTableIterator::SeekForPrev(const Slice& target) {
  }

  SavePrevIndexValue();
-  ResetBlockCacheLookupVar();

  // Call Seek() rather than SeekForPrev() in the index block, because the
  // target data block will likely to contain the position for `target`, the
@@ -207,6 +206,7 @@ void BlockBasedTableIterator::SeekForPrev(const Slice& target) {
  // to distinguish the two unless we read the second block. In this case, we'll
  // end up with reading two blocks.
  index_iter_->Seek(target);
+  is_index_at_curr_block_ = true;

  if (!index_iter_->Valid()) {
    auto seek_status = index_iter_->status();
@@ -231,8 +231,6 @@ void BlockBasedTableIterator::SeekForPrev(const Slice& target) {
    }
  }

-  is_index_at_curr_block_ = true;
-
  InitDataBlock();

  block_iter_.SeekForPrev(target);
@@ -244,22 +242,22 @@ void BlockBasedTableIterator::SeekForPrev(const Slice& target) {
 }

 void BlockBasedTableIterator::SeekToLast() {
+  direction_ = IterDirection::kBackward;
+  ResetBlockCacheLookupVar();
  is_out_of_bound_ = false;
  is_at_first_key_from_index_ = false;
  seek_stat_state_ = kNone;

  SavePrevIndexValue();
-  ResetBlockCacheLookupVar();

  index_iter_->SeekToLast();
+  is_index_at_curr_block_ = true;

  if (!index_iter_->Valid()) {
    ResetDataIter();
    return;
  }

-  is_index_at_curr_block_ = true;
-
  InitDataBlock();
  block_iter_.SeekToLast();
  FindKeyBackward();
@@ -320,7 +318,7 @@ void BlockBasedTableIterator::InitDataBlock() {
  bool use_block_cache_for_lookup = true;

  if (DoesContainBlockHandles()) {
-    data_block_handle = block_handles_.front().index_val_.handle;
+    data_block_handle = block_handles_.front().handle_;
    is_in_cache = block_handles_.front().is_cache_hit_;
    use_block_cache_for_lookup = false;
  } else {
@@ -485,15 +483,15 @@ bool BlockBasedTableIterator::MaterializeCurrentBlock() {
  // handles placed in blockhandle. So index_ will be pointing to current block.
  // After InitDataBlock, index_iter_ can point to different block if
  // BlockCacheLookupForReadAheadSize is called.
-  IndexValue index_val;
+  Slice first_internal_key;
  if (DoesContainBlockHandles()) {
-    index_val = block_handles_.front().index_val_;
+    first_internal_key = block_handles_.front().first_internal_key_;
  } else {
-    index_val = index_iter_->value();
+    first_internal_key = index_iter_->value().first_internal_key;
  }

  if (!block_iter_.Valid() ||
-      icomp_.Compare(block_iter_.key(), index_val.first_internal_key) != 0) {
+      icomp_.Compare(block_iter_.key(), first_internal_key) != 0) {
    block_iter_.Invalidate(Status::Corruption(
        "first key in index doesn't match first key in block"));
    return false;
@@ -528,7 +526,7 @@ void BlockBasedTableIterator::FindBlockForward() {
    //  index_iter_ can point to different block in case of
    //  readahead_cache_lookup_. readahead_cache_lookup_ will be handle the
    //  upper_bound check.
-    const bool next_block_is_out_of_bound =
+    bool next_block_is_out_of_bound =
        IsIndexAtCurr() && read_options_.iterate_upper_bound != nullptr &&
        block_iter_points_to_real_block_ &&
        block_upper_bound_check_ == BlockUpperBound::kUpperBoundInCurBlock;
@@ -553,8 +551,14 @@ void BlockBasedTableIterator::FindBlockForward() {
      // 2. If block_handles is empty and index is not at current because of
      //    lookup (during Next), it should skip doing index_iter_->Next(), as
      //    it's already pointing to next block;
-      if (IsIndexAtCurr()) {
+      // 3. Last block could be out of bound and it won't iterate over that
+      // during BlockCacheLookup. We need to set for that block here.
+      if (IsIndexAtCurr() || is_index_out_of_bound_) {
        index_iter_->Next();
+        if (is_index_out_of_bound_) {
+          next_block_is_out_of_bound = is_index_out_of_bound_;
+          is_index_out_of_bound_ = false;
+        }
      } else {
        // Skip Next as index_iter_ already points to correct index when it
        // iterates in BlockCacheLookupForReadAheadSize.
@@ -612,7 +616,7 @@ void BlockBasedTableIterator::FindKeyBackward() {
 }

 void BlockBasedTableIterator::CheckOutOfBound() {
-  if (IsIndexAtCurr() && read_options_.iterate_upper_bound != nullptr &&
+  if (read_options_.iterate_upper_bound != nullptr &&
      block_upper_bound_check_ != BlockUpperBound::kUpperBoundBeyondCurBlock &&
      Valid()) {
    is_out_of_bound_ =
@@ -686,13 +690,20 @@ void BlockBasedTableIterator::BlockCacheLookupForReadAheadSize(
    return;
  }

+  if (IsNextBlockOutOfBound()) {
+    updated_readahead_size = 0;
+    return;
+  }
+
  size_t current_readahead_size = 0;
  size_t footer = table_->get_rep()->footer.GetBlockTrailerSize();

  // Add the current block to block_handles_.
  {
    BlockHandleInfo block_handle_info;
-    block_handle_info.index_val_ = index_iter_->value();
+    block_handle_info.handle_ = index_iter_->value().handle;
+    block_handle_info.SetFirstInternalKey(
+        index_iter_->value().first_internal_key);
    block_handles_.emplace_back(std::move(block_handle_info));
  }

@@ -717,7 +728,9 @@ void BlockBasedTableIterator::BlockCacheLookupForReadAheadSize(
    // For current data block, do the lookup in the cache. Lookup should pin the
    // data block and add the placeholder for cache.
    BlockHandleInfo block_handle_info;
-    block_handle_info.index_val_ = index_iter_->value();
+    block_handle_info.handle_ = index_iter_->value().handle;
+    block_handle_info.SetFirstInternalKey(
+        index_iter_->value().first_internal_key);

    Status s = table_->LookupAndPinBlocksInCache<Block_kData>(
        read_options_, block_handle,
@@ -725,6 +738,7 @@ void BlockBasedTableIterator::BlockCacheLookupForReadAheadSize(
    if (!s.ok()) {
      break;
    }
+
    block_handle_info.is_cache_hit_ =
        (block_handle_info.cachable_entry_.GetValue() ||
         block_handle_info.cachable_entry_.GetCacheHandle());
@@ -738,6 +752,7 @@ void BlockBasedTableIterator::BlockCacheLookupForReadAheadSize(
    // means all the keys in next block or above are out of
    // bound.
    if (IsNextBlockOutOfBound()) {
+      is_index_out_of_bound_ = true;
      break;
    }
    index_iter_->Next();
@@ -747,10 +762,9 @@ void BlockBasedTableIterator::BlockCacheLookupForReadAheadSize(
  // update the readahead_size.
  for (auto it = block_handles_.rbegin();
       it != block_handles_.rend() && (*it).is_cache_hit_ == true; ++it) {
-    current_readahead_size -= (*it).index_val_.handle.size();
+    current_readahead_size -= (*it).handle_.size();
    current_readahead_size -= footer;
  }
-
  updated_readahead_size = current_readahead_size;
  ResetPreviousBlockOffset();
 }
@@ -250,11 +250,21 @@ class BlockBasedTableIterator : public InternalIteratorBase<Slice> {
  // BlockHandleInfo is used to store the info needed when block cache lookup
  // ahead is enabled to tune readahead_size.
  struct BlockHandleInfo {
-    BlockHandleInfo() {}
+    void SetFirstInternalKey(const Slice& key) {
+      if (key.empty()) {
+        return;
+      }
+      size_t size = key.size();
+      buf_ = std::unique_ptr<char[]>(new char[size]);
+      memcpy(buf_.get(), key.data(), size);
+      first_internal_key_ = Slice(buf_.get(), size);
+    }

-    IndexValue index_val_;
+    BlockHandle handle_;
    bool is_cache_hit_ = false;
    CachableEntry<Block> cachable_entry_;
+    Slice first_internal_key_;
+    std::unique_ptr<char[]> buf_;
  };

  bool IsIndexAtCurr() const { return is_index_at_curr_block_; }
@@ -307,6 +317,13 @@ class BlockBasedTableIterator : public InternalIteratorBase<Slice> {
  // can point to a different block. is_index_at_curr_block_ keeps track of
  // that.
  bool is_index_at_curr_block_ = true;
+  bool is_index_out_of_bound_ = false;
+
+  // Used in case of auto_readahead_size to disable the block_cache lookup if
+  // direction is reversed from forward to backward. In case of backward
+  // direction, SeekForPrev or Prev might call Seek from db_iter. So direction
+  // is used to disable the lookup.
+  IterDirection direction_ = IterDirection::kForward;

  // If `target` is null, seek to first.
  void SeekImpl(const Slice* target, bool async_prefetch);
@@ -356,6 +373,7 @@ class BlockBasedTableIterator : public InternalIteratorBase<Slice> {
                                        size_t& updated_readahead_size);

  void ResetBlockCacheLookupVar() {
+    is_index_out_of_bound_ = false;
    readahead_cache_lookup_ = false;
    ClearBlockHandles();
  }
@@ -67,9 +67,13 @@ TBlockIter* BlockBasedTable::NewDataBlockIterator(
    // might already be under way and this would invalidate it. Also, the
    // uncompression dict is typically at the end of the file and would
    // most likely break the sequentiality of the access pattern.
+    // Same is with auto_readahead_size. It iterates over index to lookup for
+    // data blocks. And this could break the the sequentiality of the access
+    // pattern.
    s = rep_->uncompression_dict_reader->GetOrReadUncompressionDictionary(
-        ro.async_io ? nullptr : prefetch_buffer, ro, no_io, ro.verify_checksums,
-        get_context, lookup_context, &uncompression_dict);
+        ((ro.async_io || ro.auto_readahead_size) ? nullptr : prefetch_buffer),
+        ro, no_io, ro.verify_checksums, get_context, lookup_context,
+        &uncompression_dict);
    if (!s.ok()) {
      iter->Invalidate(s);
      return iter;
@@ -20,7 +20,7 @@ void BlockPrefetcher::PrefetchIfNeeded(
  const size_t len = BlockBasedTable::BlockSizeWithTrailer(handle);
  const size_t offset = handle.offset();
  if (is_for_compaction) {
-    if (!rep->file->use_direct_io()) {
+    if (!rep->file->use_direct_io() && compaction_readahead_size_ > 0) {
      // If FS supports prefetching (readahead_limit_ will be non zero in that
      // case) and current block exists in prefetch buffer then return.
      if (offset + len <= readahead_limit_) {
@@ -35,11 +35,12 @@ void BlockPrefetcher::PrefetchIfNeeded(
      if (s.ok()) {
        readahead_limit_ = offset + len + compaction_readahead_size_;
        return;
+      } else if (!s.IsNotSupported()) {
+        return;
      }
    }
    // If FS prefetch is not supported, fall back to use internal prefetch
-    // buffer. Discarding other return status of Prefetch calls intentionally,
-    // as we can fallback to reading from disk if Prefetch fails.
+    // buffer.
    //
    // num_file_reads is used  by FilePrefetchBuffer only when
    // implicit_auto_readahead is set.
@@ -122,8 +123,6 @@ void BlockPrefetcher::PrefetchIfNeeded(
  }

  // If prefetch is not supported, fall back to use internal prefetch buffer.
-  // Discarding other return status of Prefetch calls intentionally, as
-  // we can fallback to reading from disk if Prefetch fails.
  IOOptions opts;
  Status s = rep->file->PrepareIOOptions(read_options, opts);
  if (!s.ok()) {
@@ -43,7 +43,7 @@ default_params = {
        [random.randint(0, 19), random.lognormvariate(2.3, 1.3)]
    ),
    "cache_index_and_filter_blocks": lambda: random.randint(0, 1),
-    "cache_size": 8388608,
+    "cache_size": lambda: random.choice([8388608, 33554432]),
    "charge_compression_dictionary_building_buffer": lambda: random.choice([0, 1]),
    "charge_filter_construction": lambda: random.choice([0, 1]),
    "charge_table_reader": lambda: random.choice([0, 1]),
@@ -126,7 +126,9 @@ default_params = {
    "mock_direct_io": False,
    "cache_type": lambda: random.choice(
        ["lru_cache", "fixed_hyper_clock_cache", "auto_hyper_clock_cache",
-         "auto_hyper_clock_cache"]
+         "auto_hyper_clock_cache", "tiered_lru_cache",
+         "tiered_fixed_hyper_clock_cache", "tiered_auto_hyper_clock_cache",
+         "tiered_auto_hyper_clock_cache"]
    ),
    "use_full_merge_v1": lambda: random.randint(0, 1),
    "use_merge": lambda: random.randint(0, 1),
@@ -163,6 +165,7 @@ default_params = {
    "db_write_buffer_size": lambda: random.choice(
        [0, 0, 0, 1024 * 1024, 8 * 1024 * 1024, 128 * 1024 * 1024]
    ),
+    "use_write_buffer_manager": lambda: random.randint(0,1),
    "avoid_unnecessary_blocking_io": random.randint(0, 1),
    "write_dbid_to_manifest": random.randint(0, 1),
    "avoid_flush_during_recovery": lambda: random.choice(
@@ -191,6 +194,7 @@ default_params = {
    ),
    "user_timestamp_size": 0,
    "secondary_cache_fault_one_in": lambda: random.choice([0, 0, 32]),
+    "compressed_secondary_cache_size": lambda: random.choice([8388608, 16777216]),
    "prepopulate_block_cache": lambda: random.choice([0, 1]),
    "memtable_prefix_bloom_size_ratio": lambda: random.choice([0.001, 0.01, 0.1, 0.5]),
    "memtable_whole_key_filtering": lambda: random.randint(0, 1),
@@ -202,7 +206,8 @@ default_params = {
    "secondary_cache_uri": lambda: random.choice(
        [
            "",
-            "compressed_secondary_cache://capacity=8388608",
+            "",
+            "",
            "compressed_secondary_cache://capacity=8388608;enable_custom_split_merge=true",
        ]
    ),
@@ -217,7 +222,7 @@ default_params = {
    "memtable_max_range_deletions": lambda: random.choice([0] * 6 + [100, 1000]),
    # 0 (disable) is the default and more commonly used value.
    "bottommost_file_compaction_delay": lambda: random.choice([0, 0, 0, 600, 3600, 86400]),
-    "auto_readahead_size" : lambda: random.choice([0, 1]),
+    "auto_readahead_size" : 0,
 }

 _TEST_DIR_ENV_VAR = "TEST_TMPDIR"
@@ -682,6 +687,22 @@ def finalize_and_sanitize(src_params):
    if dest_params["write_fault_one_in"] > 0:
        # background work may be disabled while DB is resuming after some error
        dest_params["max_write_buffer_number"] = max(dest_params["max_write_buffer_number"], 10)
+    if dest_params["secondary_cache_uri"].find("compressed_secondary_cache") >= 0:
+        dest_params["compressed_secondary_cache_size"] = 0
+        dest_params["compressed_secondary_cache_ratio"] = 0.0
+    if dest_params["cache_type"].find("tiered_") >= 0:
+        if dest_params["compressed_secondary_cache_size"] > 0:
+            dest_params["compressed_secondary_cache_ratio"] = \
+                    float(dest_params["compressed_secondary_cache_size"]/ \
+                    (dest_params["cache_size"] + dest_params["compressed_secondary_cache_size"]))
+            dest_params["compressed_secondary_cache_size"] = 0
+        else:
+            dest_params["compressed_secondary_cache_ratio"] = 0.0
+            dest_params["cache_type"] = dest_params["cache_type"].replace("tiered_", "")
+    if dest_params["use_write_buffer_manager"]:
+        if (dest_params["cache_size"] <= 0
+            or dest_params["db_write_buffer_size"] <= 0):
+            dest_params["use_write_buffer_manager"] = 0
    return dest_params


@@ -1 +0,0 @@
-Compaction read performance will regress when `Options::compaction_readahead_size` is explicitly set to 0
@@ -1 +0,0 @@
-Universal size amp compaction will conditionally exclude some of the newest L0 files when selecting input with a small negative impact to size amp. This is to prevent a large number of L0 files from being locked by a size amp compaction, potentially leading to write stop with a few more flushes.
@@ -1 +0,0 @@
-Change ldb scan command delimiter from ':' to '==>'.
@@ -1 +0,0 @@
-* Fix a bug where if there is an error reading from offset 0 of a file from L1+ and that the file is not the first file in the sorted run, data can be lost in compaction and read/scan can return incorrect results.
@@ -1 +0,0 @@
-* Fix a bug where iterator may return incorrect result for DeleteRange() users if there was an error reading from a file.
@@ -1 +0,0 @@
-* Fix a bug with atomic_flush=true that can cause DB to stuck after a flush fails (#11872). 
@@ -1 +0,0 @@
-* Fix a bug where RocksDB (with atomic_flush=false) can delete output SST files of pending flushes when a previous concurrent flush fails (#11865). This can result in DB entering read-only state with error message like `IO error: No such file or directory: While open a file for random read: /tmp/rocksdbtest-501/db_flush_test_87732_4230653031040984171/000013.sst`.  
@@ -1 +0,0 @@
-Fix an assertion fault during seek with async_io when readahead trimming is enabled.
@@ -1 +0,0 @@
-When the compressed secondary cache capacity is reduced to 0, it should be completely disabled. Before this fix, inserts and lookups would still go to the backing `LRUCache` before returning, thus incurring locking overhead. With this fix, inserts and lookups are no-ops and do not add any overhead.
@@ -1 +0,0 @@
-Updating the tiered cache (cache allocated using NewTieredCache()) by calling SetCapacity() on it was not working properly. The initial creation would set the primary cache capacity to the combined primary and compressed secondary cache capacity. But SetCapacity() would just set the primary cache capacity. With this fix, the user always specifies the total budget and compressed secondary cache ratio on creation. Subsequently, SetCapacity() will distribute the new capacity across the two caches by the same ratio. 
@@ -1 +0,0 @@
-Fixed a bug in `MultiGet` for cleaning up SuperVersion acquired with locking db mutex.
@@ -1 +0,0 @@
-* Fix a bug where row cache can falsely return kNotFound even though row cache entry is hit. 
@@ -1 +0,0 @@
-Fixed a race condition in `GenericRateLimiter` that could cause it to stop granting requests
@@ -1 +0,0 @@
-* Fix a bug (Issue #10257) where DB can hang after write stall since no compaction is scheduled (#11764).
@@ -1 +0,0 @@
-Add a fix for async_io where during seek, when reading a block for seeking a target key in a file without any readahead, the iterator aligned the read on a page boundary and reading more than necessary. This increased the storage read bandwidth usage.
@@ -1 +0,0 @@
-Fix an issue in sst dump tool to handle bounds specified for data with user-defined timestamps.
@@ -1 +0,0 @@
-* When auto_readahead_size is enabled, update readahead upper bound during readahead trimming when reseek changes iterate_upper_bound dynamically. 
@@ -1 +0,0 @@
-Fixed a bug where `rocksdb.file.read.verify.file.checksums.micros` is not populated
@@ -1 +0,0 @@
-Added an experimental new "automatic" variant of HyperClockCache that does not require a prior estimate of the average size of cache entries. This variant is activated when HyperClockCacheOptions::estimated\_entry\_charge = 0 and has essentially the same concurrency benefits as the existing HyperClockCache.
@@ -1 +0,0 @@
-* Add a new statistic `COMPACTION_CPU_TOTAL_TIME` that records cumulative compaction cpu time. This ticker is updated regularly while a compaction is running. 
@@ -1 +0,0 @@
-Add `GetEntity()` API for ReadOnly DB and Secondary DB.
@@ -1 +0,0 @@
-Add a new iterator API `Iterator::Refresh(const Snapshot *)` that allows iterator to be refreshed while using the input snapshot to read.
@@ -1 +0,0 @@
-Added a new read option `merge_operand_count_threshold`. When the number of merge operands applied during a successful point lookup exceeds this threshold, the query will return a special OK status with a new subcode `kMergeOperandThresholdExceeded`. Applications might use this signal to take action to reduce the number of merge operands for the affected key(s), for example by running a compaction.
@@ -1 +0,0 @@
-For `NewRibbonFilterPolicy()`, made the `bloom_before_level` option mutable through the Configurable interface and the SetOptions API, allowing dynamic switching between all-Bloom and all-Ribbon configurations, and configurations in between. See comments on `NewRibbonFilterPolicy()`
@@ -1 +0,0 @@
-RocksDB now allows the block cache to be stacked on top of a compressed secondary cache and a non-volatile secondary cache, thus creating a three-tier cache. To set it up, use the `NewTieredCache()` API in rocksdb/cache.h..
@@ -1 +0,0 @@
-Added a new wide-column aware full merge API called `FullMergeV3` to `MergeOperator`. `FullMergeV3` supports wide columns both as base value and merge result, which enables the application to perform more general transformations during merges. For backward compatibility, the default implementation implements the earlier logic of applying the merge operation to the default column of any wide-column entities. Specifically, if there is no base value or the base value is a plain key-value, the default implementation falls back to `FullMergeV2`. If the base value is a wide-column entity, the default implementation invokes `FullMergeV2` to perform the merge on the default column, and leaves any other columns unchanged.
@@ -1 +0,0 @@
-Add wide column support to ldb commands (scan, dump, idump, dump_wal) and sst_dump tool's scan command
@@ -1 +0,0 @@
-Added additional improvements in tuning readahead_size during Scans when auto_readahead_size is enabled. However it's not supported with Iterator::Prev operation and will return NotSupported error.
@@ -1 +0,0 @@
-During async_io, the Seek happens in 2 phases. Phase 1 starts an asynchronous read on a block cache miss, and phase 2 waits for it to complete and finishes the seek. In both phases, it tries to lookup the block cache for the data block first before looking in the prefetch buffer. It's optimized by doing the block cache lookup only in the first phase that would save some CPU.
@@ -1 +0,0 @@
-Expose more information about input files used in table creation (if any) in `CompactionFilter::Context`. See `CompactionFilter::Context::input_start_level`,`CompactionFilter::Context::input_table_properties` for more.
@@ -1 +0,0 @@
-`Options::compaction_readahead_size` 's default value is changed from 0 to 2MB.
@@ -1 +0,0 @@
-* When using LZ4 compression, the `acceleration` parameter is configurable by setting the negated value in `CompressionOptions::level`. For example, `CompressionOptions::level=-10` will set `acceleration=10`
@@ -1 +0,0 @@
-The `NewTieredCache` API has been changed to take the total cache capacity (inclusive of both the primary and the compressed secondary cache) and the ratio of total capacity to allocate to the compressed cache. These are specified in `TieredCacheOptions`. Any capacity specified in `LRUCacheOptions`, `HyperClockCacheOptions` and `CompressedSecondaryCacheOptions` is ignored. A new API, `UpdateTieredCache` is provided to dynamically update the total capacity, ratio of compressed cache, and admission policy.
@@ -1 +0,0 @@
-The `NewTieredVolatileCache()` API in rocksdb/cache.h has been renamed to `NewTieredCache()`.
@@ -1117,7 +1117,13 @@ static inline Function Choose_Extend() {
  }
 #elif defined(__SSE4_2__) && defined(__PCLMUL__) && !defined NO_THREEWAY_CRC32C
  // NOTE: runtime detection no longer supported on x86
-  (void)ExtendImpl<DefaultCRC32>;  // suppress unused warning
+#ifdef _MSC_VER
+#pragma warning(disable: 4551)
+#endif
+  (void)ExtendImpl<DefaultCRC32>; // suppress unused warning
+#ifdef _MSC_VER
+#pragma warning(default: 4551)
+#endif
  return crc32c_3way;
 #else
  return ExtendImpl<DefaultCRC32>;
@@ -78,6 +78,112 @@ INSTANTIATE_TEST_CASE_P(
        std::make_tuple(false, true, WRITE_PREPARED, kUnorderedWrite, true)));
 #endif  // !defined(ROCKSDB_VALGRIND_RUN) || defined(ROCKSDB_FULL_VALGRIND_RUN)

+TEST_P(TransactionTest, TestUpperBoundUponDeletion) {
+  // Reproduction from the original bug report, 11606
+  // This test does writes without snapshot validation, and then tries to create
+  // iterator later, which is unsupported in write unprepared.
+  if (txn_db_options.write_policy == WRITE_UNPREPARED) {
+    return;
+  }
+
+  WriteOptions write_options;
+  ReadOptions read_options;
+  Status s;
+
+  Transaction* txn = db->BeginTransaction(write_options);
+  ASSERT_TRUE(txn);
+
+  // Write some keys in a txn
+  s = txn->Put("2", "2");
+  ASSERT_OK(s);
+
+  s = txn->Put("1", "1");
+  ASSERT_OK(s);
+
+  s = txn->Delete("2");
+  ASSERT_OK(s);
+
+  read_options.iterate_upper_bound = new Slice("2", 1);
+  Iterator* iter = txn->GetIterator(read_options);
+  ASSERT_OK(iter->status());
+  iter->SeekToFirst();
+  while (iter->Valid()) {
+    ASSERT_EQ("1", iter->key().ToString());
+    iter->Next();
+  }
+  delete iter;
+  delete txn;
+  delete read_options.iterate_upper_bound;
+}
+
+TEST_P(TransactionTest, TestTxnRespectBoundsInReadOption) {
+  if (txn_db_options.write_policy == WRITE_UNPREPARED) {
+    return;
+  }
+
+  WriteOptions write_options;
+
+  {
+    std::unique_ptr<Transaction> txn(db->BeginTransaction(write_options));
+    // writes that should be observed by base_iterator_ in BaseDeltaIterator
+    ASSERT_OK(txn->Put("a", "aa"));
+    ASSERT_OK(txn->Put("c", "cc"));
+    ASSERT_OK(txn->Put("e", "ee"));
+    ASSERT_OK(txn->Put("f", "ff"));
+    ASSERT_TRUE(txn->Commit().ok());
+  }
+
+  std::unique_ptr<Transaction> txn2(db->BeginTransaction(write_options));
+  // writes that should be observed by delta_iterator_ in BaseDeltaIterator
+  ASSERT_OK(txn2->Put("b", "bb"));
+  ASSERT_OK(txn2->Put("c", "cc"));
+  ASSERT_OK(txn2->Put("f", "ff"));
+
+  // delta_iterator_:   b c   f
+  //  base_iterator_: a   c e f
+  //
+  // given range [c, f)
+  // assert only {c, e} can be seen
+
+  ReadOptions ro;
+  ro.iterate_lower_bound = new Slice("c");
+  ro.iterate_upper_bound = new Slice("f");
+  std::unique_ptr<Iterator> iter(txn2->GetIterator(ro));
+
+  iter->Seek(Slice("b"));
+  ASSERT_EQ("c", iter->key());  // lower bound capping
+  iter->Seek(Slice("f"));
+  ASSERT_FALSE(iter->Valid());  // out of bound
+
+  iter->SeekForPrev(Slice("f"));
+  ASSERT_EQ("e", iter->key());  // upper bound capping
+  iter->SeekForPrev(Slice("b"));
+  ASSERT_FALSE(iter->Valid());  // out of bound
+
+  // move to the lower bound
+  iter->SeekToFirst();
+  ASSERT_EQ("c", iter->key());
+  iter->Prev();
+  ASSERT_FALSE(iter->Valid());
+
+  // move to the upper bound
+  iter->SeekToLast();
+  ASSERT_EQ("e", iter->key());
+  iter->Next();
+  ASSERT_FALSE(iter->Valid());
+
+  // reversely walk to the beginning
+  iter->SeekToLast();
+  ASSERT_EQ("e", iter->key());
+  iter->Prev();
+  ASSERT_EQ("c", iter->key());
+  iter->Prev();
+  ASSERT_FALSE(iter->Valid());
+
+  delete ro.iterate_lower_bound;
+  delete ro.iterate_upper_bound;
+}
+
 TEST_P(TransactionTest, DoubleEmptyWrite) {
  WriteOptions write_options;
  write_options.sync = true;
@@ -123,11 +123,7 @@ Status WritePreparedTxn::GetImpl(const ReadOptions& options,
 }

 Iterator* WritePreparedTxn::GetIterator(const ReadOptions& options) {
-  // Make sure to get iterator from WritePrepareTxnDB, not the root db.
-  Iterator* db_iter = wpt_db_->NewIterator(options);
-  assert(db_iter);
-
-  return write_batch_.NewIteratorWithBase(db_iter);
+  return GetIterator(options, wpt_db_->DefaultColumnFamily());
 }

 Iterator* WritePreparedTxn::GetIterator(const ReadOptions& options,
@@ -136,7 +132,7 @@ Iterator* WritePreparedTxn::GetIterator(const ReadOptions& options,
  Iterator* db_iter = wpt_db_->NewIterator(options, column_family);
  assert(db_iter);

-  return write_batch_.NewIteratorWithBase(column_family, db_iter);
+  return write_batch_.NewIteratorWithBase(column_family, db_iter, &options);
 }

 Status WritePreparedTxn::PrepareInternal() {
@@ -1037,7 +1037,8 @@ Iterator* WriteUnpreparedTxn::GetIterator(const ReadOptions& options,
  Iterator* db_iter = wupt_db_->NewIterator(options, column_family, this);
  assert(db_iter);

-  auto iter = write_batch_.NewIteratorWithBase(column_family, db_iter);
+  auto iter =
+      write_batch_.NewIteratorWithBase(column_family, db_iter, &options);
  active_iterators_.push_back(iter);
  iter->RegisterCleanup(CleanupWriteUnpreparedWBWIIterator, this, iter);
  return iter;
@@ -298,12 +298,20 @@ WBWIIterator* WriteBatchWithIndex::NewIterator(
 Iterator* WriteBatchWithIndex::NewIteratorWithBase(
    ColumnFamilyHandle* column_family, Iterator* base_iterator,
    const ReadOptions* read_options) {
-  auto wbwiii =
-      new WBWIIteratorImpl(GetColumnFamilyID(column_family), &(rep->skip_list),
-                           &rep->write_batch, &rep->comparator);
+  WBWIIteratorImpl* wbwiii;
+  if (read_options != nullptr) {
+    wbwiii = new WBWIIteratorImpl(
+        GetColumnFamilyID(column_family), &(rep->skip_list), &rep->write_batch,
+        &rep->comparator, read_options->iterate_lower_bound,
+        read_options->iterate_upper_bound);
+  } else {
+    wbwiii = new WBWIIteratorImpl(GetColumnFamilyID(column_family),
+                                  &(rep->skip_list), &rep->write_batch,
+                                  &rep->comparator);
+  }
+
  return new BaseDeltaIterator(column_family, base_iterator, wbwiii,
-                               GetColumnFamilyUserComparator(column_family),
-                               read_options);
+                               GetColumnFamilyUserComparator(column_family));
 }

 Iterator* WriteBatchWithIndex::NewIteratorWithBase(Iterator* base_iterator) {
@@ -21,17 +21,16 @@ namespace ROCKSDB_NAMESPACE {
 BaseDeltaIterator::BaseDeltaIterator(ColumnFamilyHandle* column_family,
                                     Iterator* base_iterator,
                                     WBWIIteratorImpl* delta_iterator,
-                                     const Comparator* comparator,
-                                     const ReadOptions* read_options)
+                                     const Comparator* comparator)
    : forward_(true),
      current_at_base_(true),
      equal_keys_(false),
      status_(Status::OK()),
      base_iterator_(base_iterator),
      delta_iterator_(delta_iterator),
-      comparator_(comparator),
-      iterate_upper_bound_(read_options ? read_options->iterate_upper_bound
-                                        : nullptr) {
+      comparator_(comparator) {
+  assert(base_iterator_);
+  assert(delta_iterator_);
  assert(comparator_);
  wbwii_.reset(new WriteBatchWithIndexInternal(column_family));
 }
@@ -304,14 +303,6 @@ void BaseDeltaIterator::UpdateCurrent() {
        // Finished
        return;
      }
-      if (iterate_upper_bound_) {
-        if (comparator_->CompareWithoutTimestamp(
-                delta_entry.key, /*a_has_ts=*/false, *iterate_upper_bound_,
-                /*b_has_ts=*/false) >= 0) {
-          // out of upper bound -> finished.
-          return;
-        }
-      }
      if (delta_result == WBWIIteratorImpl::kDeleted &&
          wbwii_->GetNumOperands() == 0) {
        AdvanceDelta();
@@ -36,8 +36,7 @@ class BaseDeltaIterator : public Iterator {
 public:
  BaseDeltaIterator(ColumnFamilyHandle* column_family, Iterator* base_iterator,
                    WBWIIteratorImpl* delta_iterator,
-                    const Comparator* comparator,
-                    const ReadOptions* read_options = nullptr);
+                    const Comparator* comparator);

  ~BaseDeltaIterator() override {}

@@ -71,7 +70,6 @@ class BaseDeltaIterator : public Iterator {
  std::unique_ptr<Iterator> base_iterator_;
  std::unique_ptr<WBWIIteratorImpl> delta_iterator_;
  const Comparator* comparator_;  // not owned
-  const Slice* iterate_upper_bound_;
  mutable PinnableSlice merge_result_;
 };

@@ -197,59 +195,107 @@ class WBWIIteratorImpl : public WBWIIterator {
  WBWIIteratorImpl(uint32_t column_family_id,
                   WriteBatchEntrySkipList* skip_list,
                   const ReadableWriteBatch* write_batch,
-                   WriteBatchEntryComparator* comparator)
+                   WriteBatchEntryComparator* comparator,
+                   const Slice* iterate_lower_bound = nullptr,
+                   const Slice* iterate_upper_bound = nullptr)
      : column_family_id_(column_family_id),
        skip_list_iter_(skip_list),
        write_batch_(write_batch),
-        comparator_(comparator) {}
+        comparator_(comparator),
+        iterate_lower_bound_(iterate_lower_bound),
+        iterate_upper_bound_(iterate_upper_bound) {}

  ~WBWIIteratorImpl() override {}

  bool Valid() const override {
-    if (!skip_list_iter_.Valid()) {
-      return false;
-    }
-    const WriteBatchIndexEntry* iter_entry = skip_list_iter_.key();
-    return (iter_entry != nullptr &&
-            iter_entry->column_family == column_family_id_);
+    return !out_of_bound_ && ValidRegardlessOfBoundLimit();
  }

  void SeekToFirst() override {
-    WriteBatchIndexEntry search_entry(
-        nullptr /* search_key */, column_family_id_,
-        true /* is_forward_direction */, true /* is_seek_to_first */);
-    skip_list_iter_.Seek(&search_entry);
+    if (iterate_lower_bound_ != nullptr) {
+      WriteBatchIndexEntry search_entry(
+          iterate_lower_bound_ /* search_key */, column_family_id_,
+          true /* is_forward_direction */, false /* is_seek_to_first */);
+      skip_list_iter_.Seek(&search_entry);
+    } else {
+      WriteBatchIndexEntry search_entry(
+          nullptr /* search_key */, column_family_id_,
+          true /* is_forward_direction */, true /* is_seek_to_first */);
+      skip_list_iter_.Seek(&search_entry);
+    }
+
+    if (ValidRegardlessOfBoundLimit()) {
+      out_of_bound_ = TestOutOfBound();
+    }
  }

  void SeekToLast() override {
-    WriteBatchIndexEntry search_entry(
-        nullptr /* search_key */, column_family_id_ + 1,
-        true /* is_forward_direction */, true /* is_seek_to_first */);
+    WriteBatchIndexEntry search_entry =
+        (iterate_upper_bound_ != nullptr)
+            ? WriteBatchIndexEntry(
+                  iterate_upper_bound_ /* search_key */, column_family_id_,
+                  true /* is_forward_direction */, false /* is_seek_to_first */)
+            : WriteBatchIndexEntry(
+                  nullptr /* search_key */, column_family_id_ + 1,
+                  true /* is_forward_direction */, true /* is_seek_to_first */);
+
    skip_list_iter_.Seek(&search_entry);
    if (!skip_list_iter_.Valid()) {
      skip_list_iter_.SeekToLast();
    } else {
      skip_list_iter_.Prev();
    }
+
+    if (ValidRegardlessOfBoundLimit()) {
+      out_of_bound_ = TestOutOfBound();
+    }
  }

  void Seek(const Slice& key) override {
+    if (BeforeLowerBound(&key)) {  // cap to prevent out of bound
+      SeekToFirst();
+      return;
+    }
+
    WriteBatchIndexEntry search_entry(&key, column_family_id_,
                                      true /* is_forward_direction */,
                                      false /* is_seek_to_first */);
    skip_list_iter_.Seek(&search_entry);
+
+    if (ValidRegardlessOfBoundLimit()) {
+      out_of_bound_ = TestOutOfBound();
+    }
  }

  void SeekForPrev(const Slice& key) override {
+    if (AtOrAfterUpperBound(&key)) {  // cap to prevent out of bound
+      SeekToLast();
+      return;
+    }
+
    WriteBatchIndexEntry search_entry(&key, column_family_id_,
                                      false /* is_forward_direction */,
                                      false /* is_seek_to_first */);
    skip_list_iter_.SeekForPrev(&search_entry);
+
+    if (ValidRegardlessOfBoundLimit()) {
+      out_of_bound_ = TestOutOfBound();
+    }
  }

-  void Next() override { skip_list_iter_.Next(); }
+  void Next() override {
+    skip_list_iter_.Next();
+    if (ValidRegardlessOfBoundLimit()) {
+      out_of_bound_ = TestOutOfBound();
+    }
+  }

-  void Prev() override { skip_list_iter_.Prev(); }
+  void Prev() override {
+    skip_list_iter_.Prev();
+    if (ValidRegardlessOfBoundLimit()) {
+      out_of_bound_ = TestOutOfBound();
+    }
+  }

  WriteEntry Entry() const override;

@@ -290,6 +336,45 @@ class WBWIIteratorImpl : public WBWIIterator {
  WriteBatchEntrySkipList::Iterator skip_list_iter_;
  const ReadableWriteBatch* write_batch_;
  WriteBatchEntryComparator* comparator_;
+  const Slice* iterate_lower_bound_;
+  const Slice* iterate_upper_bound_;
+  bool out_of_bound_ = false;
+
+  bool TestOutOfBound() const {
+    const Slice& curKey = Entry().key;
+    return AtOrAfterUpperBound(&curKey) || BeforeLowerBound(&curKey);
+  }
+
+  bool ValidRegardlessOfBoundLimit() const {
+    if (!skip_list_iter_.Valid()) {
+      return false;
+    }
+    const WriteBatchIndexEntry* iter_entry = skip_list_iter_.key();
+    return iter_entry != nullptr &&
+           iter_entry->column_family == column_family_id_;
+  }
+
+  bool AtOrAfterUpperBound(const Slice* k) const {
+    if (iterate_upper_bound_ == nullptr) {
+      return false;
+    }
+
+    return comparator_->GetComparator(column_family_id_)
+               ->CompareWithoutTimestamp(*k, /*a_has_ts=*/false,
+                                         *iterate_upper_bound_,
+                                         /*b_has_ts=*/false) >= 0;
+  }
+
+  bool BeforeLowerBound(const Slice* k) const {
+    if (iterate_lower_bound_ == nullptr) {
+      return false;
+    }
+
+    return comparator_->GetComparator(column_family_id_)
+               ->CompareWithoutTimestamp(*k, /*a_has_ts=*/false,
+                                         *iterate_lower_bound_,
+                                         /*b_has_ts=*/false) < 0;
+  }
 };

 class WriteBatchWithIndexInternal {
@@ -1641,6 +1641,104 @@ TEST_P(WriteBatchWithIndexTest, TestNewIteratorWithBaseFromWbwi) {
  ASSERT_OK(iter->status());
 }

+TEST_P(WriteBatchWithIndexTest, TestBoundsCheckingInDeltaIterator) {
+  Status s = OpenDB();
+  ASSERT_OK(s);
+
+  KVMap empty_map;
+
+  // writes that should be observed by BaseDeltaIterator::delta_iterator_
+  ASSERT_OK(batch_->Put("a", "aa"));
+  ASSERT_OK(batch_->Put("b", "bb"));
+  ASSERT_OK(batch_->Put("c", "cc"));
+
+  ReadOptions ro;
+
+  auto check_only_b_is_visible = [&]() {
+    std::unique_ptr<Iterator> iter(batch_->NewIteratorWithBase(
+        db_->DefaultColumnFamily(), new KVIter(&empty_map), &ro));
+
+    // move to the lower bound
+    iter->SeekToFirst();
+    ASSERT_EQ("b", iter->key());
+    iter->Prev();
+    ASSERT_FALSE(iter->Valid());
+
+    // move to the upper bound
+    iter->SeekToLast();
+    ASSERT_EQ("b", iter->key());
+    iter->Next();
+    ASSERT_FALSE(iter->Valid());
+
+    // test bounds checking in Seek and SeekForPrev
+    iter->Seek(Slice("a"));
+    ASSERT_EQ("b", iter->key());
+    iter->Seek(Slice("b"));
+    ASSERT_EQ("b", iter->key());
+    iter->Seek(Slice("c"));
+    ASSERT_FALSE(iter->Valid());
+
+    iter->SeekForPrev(Slice("c"));
+    ASSERT_EQ("b", iter->key());
+    iter->SeekForPrev(Slice("b"));
+    ASSERT_EQ("b", iter->key());
+    iter->SeekForPrev(Slice("a"));
+    ASSERT_FALSE(iter->Valid());
+
+    iter->SeekForPrev(
+        Slice("a.1"));  // a non-existent key that is smaller than "b"
+    ASSERT_FALSE(iter->Valid());
+
+    iter->Seek(Slice("b.1"));  // a non-existent key that is greater than "b"
+    ASSERT_FALSE(iter->Valid());
+
+    delete ro.iterate_lower_bound;
+    delete ro.iterate_upper_bound;
+  };
+
+  ro.iterate_lower_bound = new Slice("b");
+  ro.iterate_upper_bound = new Slice("c");
+  check_only_b_is_visible();
+
+  ro.iterate_lower_bound = new Slice("a.1");
+  ro.iterate_upper_bound = new Slice("c");
+  check_only_b_is_visible();
+
+  ro.iterate_lower_bound = new Slice("b");
+  ro.iterate_upper_bound = new Slice("b.2");
+  check_only_b_is_visible();
+}
+
+TEST_P(WriteBatchWithIndexTest,
+       TestBoundsCheckingInSeekToFirstAndLastOfDeltaIterator) {
+  Status s = OpenDB();
+  ASSERT_OK(s);
+  KVMap empty_map;
+  // writes that should be observed by BaseDeltaIterator::delta_iterator_
+  ASSERT_OK(batch_->Put("c", "cc"));
+
+  ReadOptions ro;
+  auto check_nothing_visible = [&]() {
+    std::unique_ptr<Iterator> iter(batch_->NewIteratorWithBase(
+        db_->DefaultColumnFamily(), new KVIter(&empty_map), &ro));
+    iter->SeekToFirst();
+    ASSERT_FALSE(iter->Valid());
+    iter->SeekToLast();
+    ASSERT_FALSE(iter->Valid());
+
+    delete ro.iterate_lower_bound;
+    delete ro.iterate_upper_bound;
+  };
+
+  ro.iterate_lower_bound = new Slice("b");
+  ro.iterate_upper_bound = new Slice("c");
+  check_nothing_visible();
+
+  ro.iterate_lower_bound = new Slice("d");
+  ro.iterate_upper_bound = new Slice("e");
+  check_nothing_visible();
+}
+
 TEST_P(WriteBatchWithIndexTest, SavePointTest) {
  ColumnFamilyHandleImplDummy cf1(1, BytewiseComparator());
  KVMap empty_map;
				`@@ -1 +0,0 @@`
				Compaction read performance will regress when `Options::compaction_readahead_size` is explicitly set to 0
				`@@ -1 +0,0 @@`
				`Universal size amp compaction will conditionally exclude some of the newest L0 files when selecting input with a small negative impact to size amp. This is to prevent a large number of L0 files from being locked by a size amp compaction, potentially leading to write stop with a few more flushes.`
				`@@ -1 +0,0 @@`
				`Change ldb scan command delimiter from ':' to '==>'.`
				`@@ -1 +0,0 @@`
				`* Fix a bug where if there is an error reading from offset 0 of a file from L1+ and that the file is not the first file in the sorted run, data can be lost in compaction and read/scan can return incorrect results.`
				`@@ -1 +0,0 @@`
				`* Fix a bug where iterator may return incorrect result for DeleteRange() users if there was an error reading from a file.`
				`@@ -1 +0,0 @@`
				`* Fix a bug with atomic_flush=true that can cause DB to stuck after a flush fails (#11872).`
				`@@ -1 +0,0 @@`
				* Fix a bug where RocksDB (with atomic_flush=false) can delete output SST files of pending flushes when a previous concurrent flush fails (#11865). This can result in DB entering read-only state with error message like `IO error: No such file or directory: While open a file for random read: /tmp/rocksdbtest-501/db_flush_test_87732_4230653031040984171/000013.sst`.
				`@@ -1 +0,0 @@`
				`Fix an assertion fault during seek with async_io when readahead trimming is enabled.`
				`@@ -1 +0,0 @@`
				When the compressed secondary cache capacity is reduced to 0, it should be completely disabled. Before this fix, inserts and lookups would still go to the backing `LRUCache` before returning, thus incurring locking overhead. With this fix, inserts and lookups are no-ops and do not add any overhead.
				`@@ -1 +0,0 @@`
				Updating the tiered cache (cache allocated using NewTieredCache()) by calling SetCapacity() on it was not working properly. The initial creation would set the primary cache capacity to the combined primary and compressed secondary cache capacity. But SetCapacity() would just set the primary cache capacity. With this fix, the user always specifies the total budget and compressed secondary cache ratio on creation. Subsequently, SetCapacity() will distribute the new capacity across the two caches by the same ratio.