Update HISTORY and version.h for 8.6.7

Summary:
**Context/Summary:**
https://github.com/facebook/rocksdb/pull/11631 introduced an undesired fallback behavior to RocksDB internal prefetching even when FS prefetching return non-OK status other than "Unsupported". We only want to fall back when FS prefetching is not supported.

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

Test Plan: CI

Reviewed By: ajkr

Differential Revision: D49667055

Pulled By: hx235

fbshipit-source-id: fa36e4e5d6dc9507080217035f9d6ff8e4abda28

HISTORY.md

@@ -1,6 +1,68 @@
 # Rocksdb Change Log
 > NOTE: Entries for next release do not go here. Follow instructions in `unreleased_history/README.txt`
 
+## 8.6.7 (09/26/2023)
+### Bug Fixes
+* 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()`
+
+### Behavior Changes
+* For non direct IO, eliminate the file system prefetching attempt for compaction read when `Options::compaction_readahead_size` is 0
+
+## 8.6.6 (09/25/2023)
+### Bug Fixes
+* 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`.
+* 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.
+
+## 8.6.5 (09/15/2023)
+### Bug Fixes
+* Fixed a bug where `rocksdb.file.read.verify.file.checksums.micros` is not populated.
+
+## 8.6.4 (09/13/2023)
+### Public API changes
+* Add a column family option `default_temperature` that is used for file reading accounting purpose, such as io statistics, for files that don't have an explicitly set temperature.
+
+## 8.6.3 (09/12/2023)
+### 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.
+
+## 8.6.2 (09/11/2023)
+### Bug Fixes
+* 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.
+
+## 8.6.1 (08/30/2023)
+### Public API Changes
+* `Options::compaction_readahead_size` 's default value is changed from 0 to 2MB.
+
+### Behavior Changes
+* Compaction read performance will regress when `Options::compaction_readahead_size` is explicitly set to 0
+
+## 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.
+* Add a new feature to trim readahead_size during scans upto upper_bound when iterate_upper_bound is specified. It's enabled through ReadOptions.auto_readahead_size. Users must also specify ReadOptions.iterate_upper_bound.
+* RocksDB will compare the number of input keys to the number of keys processed after each compaction. Compaction will fail and report Corruption status if the verification fails. Option `compaction_verify_record_count` is introduced for this purpose and is enabled by default.
+* Add a CF option `bottommost_file_compaction_delay` to allow specifying the delay of bottommost level single-file compactions.
+* Add support to allow enabling / disabling user-defined timestamps feature for an existing column family in combination with the in-Memtable only feature.
+* Implement a new admission policy for the compressed secondary cache that admits blocks evicted from the primary cache with the hit bit set. This policy can be specified in TieredVolatileCacheOptions by setting the newly added adm_policy option.
+* Add a column family option `memtable_max_range_deletions` that limits the number of range deletions in a memtable. RocksDB will try to do an automatic flush after the limit is reached. (#11358)
+* Add PutEntity API in sst_file_writer
+* Add `timeout` in microsecond option to `WaitForCompactOptions` to allow timely termination of prolonged waiting in scenarios like recurring recoverable errors, such as out-of-space situations and continuous write streams that sustain ongoing flush and compactions
+* New statistics `rocksdb.file.read.{get|multiget|db.iterator|verify.checksum|verify.file.checksums}.micros` measure read time of block-based SST tables or blob files during db open, `Get()`, `MultiGet()`, using db iterator, `VerifyFileChecksums()` and `VerifyChecksum()`. They require stats level greater than `StatsLevel::kExceptDetailedTimers`.
+* Add close_db option to `WaitForCompactOptions` to call Close() after waiting is done.
+* Add a new compression option `CompressionOptions::checksum` for enabling ZSTD's checksum feature to detect corruption during decompression.
+
+### Public API Changes
+* Mark `Options::access_hint_on_compaction_start` related APIs as deprecated. See #11631 for alternative behavior.
+
+### Behavior Changes
+* Statistics `rocksdb.sst.read.micros` now includes time spent on multi read and async read into the file
+* For Universal Compaction users, periodic compaction (option `periodic_compaction_seconds`) will be set to 30 days by default if block based table is used.
+
+### Bug Fixes
+* Fix a bug in FileTTLBooster that can cause users with a large number of levels (more than 65) to see errors like "runtime error: shift exponent .. is too large.." (#11673).
+
 ## 8.5.0 (07/21/2023)
 ### Public API Changes
 * Removed recently added APIs `GeneralCache` and `MakeSharedGeneralCache()` as our plan changed to stop exposing a general-purpose cache interface. The old forms of these APIs, `Cache` and `NewLRUCache()`, are still available, although general-purpose caching support will be dropped eventually.

cache/compressed_secondary_cache.cc

@@ -22,7 +22,8 @@ CompressedSecondaryCache::CompressedSecondaryCache(
       cache_options_(opts),
       cache_res_mgr_(std::make_shared<ConcurrentCacheReservationManager>(
           std::make_shared<CacheReservationManagerImpl<CacheEntryRole::kMisc>>(
-              cache_))) {}
+              cache_))),
+      disable_cache_(opts.capacity == 0) {}
 
 CompressedSecondaryCache::~CompressedSecondaryCache() {
   assert(cache_res_mgr_->GetTotalReservedCacheSize() == 0);
@@ -33,6 +34,10 @@ std::unique_ptr<SecondaryCacheResultHandle> CompressedSecondaryCache::Lookup(
     Cache::CreateContext* create_context, bool /*wait*/, bool advise_erase,
     bool& kept_in_sec_cache) {
   assert(helper);
+  if (disable_cache_) {
+    return nullptr;
+  }
+
   std::unique_ptr<SecondaryCacheResultHandle> handle;
   kept_in_sec_cache = false;
   Cache::Handle* lru_handle = cache_->Lookup(key);
@@ -115,6 +120,10 @@ Status CompressedSecondaryCache::Insert(const Slice& key,
     return Status::InvalidArgument();
   }
 
+  if (disable_cache_) {
+    return Status::OK();
+  }
+
   auto internal_helper = GetHelper(cache_options_.enable_custom_split_merge);
   if (!force_insert) {
     Cache::Handle* lru_handle = cache_->Lookup(key);
@@ -186,6 +195,7 @@ Status CompressedSecondaryCache::SetCapacity(size_t capacity) {
   MutexLock l(&capacity_mutex_);
   cache_options_.capacity = capacity;
   cache_->SetCapacity(capacity);
+  disable_cache_ = capacity == 0;
   return Status::OK();
 }
 

cache/compressed_secondary_cache.h

@@ -136,6 +136,7 @@ class CompressedSecondaryCache : public SecondaryCache {
   CompressedSecondaryCacheOptions cache_options_;
   mutable port::Mutex capacity_mutex_;
   std::shared_ptr<ConcurrentCacheReservationManager> cache_res_mgr_;
+  bool disable_cache_;
 };
 
 }  // namespace ROCKSDB_NAMESPACE

db/db_compaction_test.cc

@@ -6025,23 +6025,18 @@ TEST_P(DBCompactionDirectIOTest, DirectIO) {
   options.use_direct_io_for_flush_and_compaction = GetParam();
   options.env = MockEnv::Create(Env::Default());
   Reopen(options);
-  bool readahead = false;
   SyncPoint::GetInstance()->SetCallBack(
       "CompactionJob::OpenCompactionOutputFile", [&](void* arg) {
         bool* use_direct_writes = static_cast<bool*>(arg);
         ASSERT_EQ(*use_direct_writes,
                   options.use_direct_io_for_flush_and_compaction);
       });
-  if (options.use_direct_io_for_flush_and_compaction) {
-    SyncPoint::GetInstance()->SetCallBack(
-        "SanitizeOptions:direct_io", [&](void* /*arg*/) { readahead = true; });
-  }
   SyncPoint::GetInstance()->EnableProcessing();
   CreateAndReopenWithCF({"pikachu"}, options);
   MakeTables(3, "p", "q", 1);
   ASSERT_EQ("1,1,1", FilesPerLevel(1));
   Compact(1, "p", "q");
-  ASSERT_EQ(readahead, options.use_direct_reads);
+  ASSERT_EQ(false, options.use_direct_reads);
   ASSERT_EQ("0,0,1", FilesPerLevel(1));
   Destroy(options);
   delete options.env;

db/db_flush_test.cc

@@ -3193,6 +3193,279 @@ INSTANTIATE_TEST_CASE_P(DBFlushDirectIOTest, DBFlushDirectIOTest,
 
 INSTANTIATE_TEST_CASE_P(DBAtomicFlushTest, DBAtomicFlushTest, testing::Bool());
 
+TEST_F(DBFlushTest, NonAtomicFlushRollbackPendingFlushes) {
+  // Fix a bug in when atomic_flush=false.
+  // The bug can happen as follows:
+  // Start Flush0 for memtable M0 to SST0
+  // Start Flush1 for memtable M1 to SST1
+  // Flush1 returns OK, but don't install to MANIFEST and let whoever flushes
+  // M0 to take care of it
+  // Flush0 finishes with a retryable IOError
+  //   - It rollbacks M0, (incorrectly) not M1
+  //   - Deletes SST1 and SST2
+  //
+  // Auto-recovery will start Flush2 for M0, it does not pick up M1 since it
+  // thinks that M1 is flushed
+  // Flush2 writes SST3 and finishes OK, tries to install SST3 and SST2
+  // Error opening SST2 since it's already deleted
+  //
+  // The fix is to let Flush0 also rollback M1.
+  Options opts = CurrentOptions();
+  opts.atomic_flush = false;
+  opts.memtable_factory.reset(test::NewSpecialSkipListFactory(1));
+  opts.max_write_buffer_number = 64;
+  opts.max_background_flushes = 4;
+  env_->SetBackgroundThreads(4, Env::HIGH);
+  DestroyAndReopen(opts);
+  std::atomic_int flush_count = 0;
+  SyncPoint::GetInstance()->ClearAllCallBacks();
+  SyncPoint::GetInstance()->DisableProcessing();
+  SyncPoint::GetInstance()->SetCallBack(
+      "FlushJob::WriteLevel0Table:s", [&](void* s_ptr) {
+        int c = flush_count.fetch_add(1);
+        if (c == 0) {
+          Status* s = (Status*)(s_ptr);
+          IOStatus io_error = IOStatus::IOError("injected foobar");
+          io_error.SetRetryable(true);
+          *s = io_error;
+          TEST_SYNC_POINT("Let mem1 flush start");
+          TEST_SYNC_POINT("Wait for mem1 flush to finish");
+        }
+      });
+  SyncPoint::GetInstance()->LoadDependency(
+      {{"Let mem1 flush start", "Mem1 flush starts"},
+       {"DBImpl::BGWorkFlush:done", "Wait for mem1 flush to finish"},
+       {"RecoverFromRetryableBGIOError:RecoverSuccess",
+        "Wait for error recover"}});
+  // Need first flush to wait for the second flush to finish
+  SyncPoint::GetInstance()->EnableProcessing();
+  ASSERT_OK(Put(Key(1), "val1"));
+  // trigger bg flush mem0
+  ASSERT_OK(Put(Key(2), "val2"));
+  TEST_SYNC_POINT("Mem1 flush starts");
+  // trigger bg flush mem1
+  ASSERT_OK(Put(Key(3), "val3"));
+
+  TEST_SYNC_POINT("Wait for error recover");
+  ASSERT_EQ(1, NumTableFilesAtLevel(0));
+  SyncPoint::GetInstance()->ClearAllCallBacks();
+  SyncPoint::GetInstance()->DisableProcessing();
+}
+
+TEST_F(DBFlushTest, AbortNonAtomicFlushWhenBGError) {
+  // Fix a bug in when atomic_flush=false.
+  // The bug can happen as follows:
+  // Start Flush0 for memtable M0 to SST0
+  // Start Flush1 for memtable M1 to SST1
+  // Flush1 returns OK, but doesn't install output MANIFEST and let whoever
+  // flushes M0 to take care of it
+  // Start Flush2 for memtable M2 to SST2
+  // Flush0 finishes with a retryable IOError
+  //   - It rollbacks M0 AND M1
+  //   - Deletes SST1 and SST2
+  // Flush2 finishes, does not rollback M2,
+  //  - releases the pending file number that keeps SST2 alive
+  //  - deletes SST2
+  //
+  // Then auto-recovery starts, error opening SST2 when try to install
+  // flush result
+  //
+  // The fix is to let Flush2 rollback M2 if it finds that
+  // there is a background error.
+  Options opts = CurrentOptions();
+  opts.atomic_flush = false;
+  opts.memtable_factory.reset(test::NewSpecialSkipListFactory(1));
+  opts.max_write_buffer_number = 64;
+  opts.max_background_flushes = 4;
+  env_->SetBackgroundThreads(4, Env::HIGH);
+  DestroyAndReopen(opts);
+  std::atomic_int flush_count = 0;
+  SyncPoint::GetInstance()->ClearAllCallBacks();
+  SyncPoint::GetInstance()->DisableProcessing();
+  SyncPoint::GetInstance()->SetCallBack(
+      "FlushJob::WriteLevel0Table:s", [&](void* s_ptr) {
+        int c = flush_count.fetch_add(1);
+        if (c == 0) {
+          Status* s = (Status*)(s_ptr);
+          IOStatus io_error = IOStatus::IOError("injected foobar");
+          io_error.SetRetryable(true);
+          *s = io_error;
+          TEST_SYNC_POINT("Let mem1 flush start");
+          TEST_SYNC_POINT("Wait for mem1 flush to finish");
+
+          TEST_SYNC_POINT("Let mem2 flush start");
+          TEST_SYNC_POINT("Wait for mem2 to start writing table");
+        }
+      });
+
+  SyncPoint::GetInstance()->SetCallBack(
+      "FlushJob::WriteLevel0Table", [&](void* mems) {
+        autovector<MemTable*>* mems_ptr = (autovector<MemTable*>*)mems;
+        if ((*mems_ptr)[0]->GetID() == 3) {
+          TEST_SYNC_POINT("Mem2 flush starts writing table");
+          TEST_SYNC_POINT("Mem2 flush waits until rollback");
+        }
+      });
+  SyncPoint::GetInstance()->LoadDependency(
+      {{"Let mem1 flush start", "Mem1 flush starts"},
+       {"DBImpl::BGWorkFlush:done", "Wait for mem1 flush to finish"},
+       {"Let mem2 flush start", "Mem2 flush starts"},
+       {"Mem2 flush starts writing table",
+        "Wait for mem2 to start writing table"},
+       {"RollbackMemtableFlush", "Mem2 flush waits until rollback"},
+       {"RecoverFromRetryableBGIOError:RecoverSuccess",
+        "Wait for error recover"}});
+  SyncPoint::GetInstance()->EnableProcessing();
+
+  ASSERT_OK(Put(Key(1), "val1"));
+  // trigger bg flush mem0
+  ASSERT_OK(Put(Key(2), "val2"));
+  TEST_SYNC_POINT("Mem1 flush starts");
+  // trigger bg flush mem1
+  ASSERT_OK(Put(Key(3), "val3"));
+
+  TEST_SYNC_POINT("Mem2 flush starts");
+  ASSERT_OK(Put(Key(4), "val4"));
+
+  TEST_SYNC_POINT("Wait for error recover");
+  // Recovery flush writes 3 memtables together into 1 file.
+  ASSERT_EQ(1, NumTableFilesAtLevel(0));
+  SyncPoint::GetInstance()->ClearAllCallBacks();
+  SyncPoint::GetInstance()->DisableProcessing();
+}
+
+TEST_F(DBFlushTest, NonAtomicNormalFlushAbortWhenBGError) {
+  Options opts = CurrentOptions();
+  opts.atomic_flush = false;
+  opts.memtable_factory.reset(test::NewSpecialSkipListFactory(1));
+  opts.max_write_buffer_number = 64;
+  opts.max_background_flushes = 1;
+  env_->SetBackgroundThreads(2, Env::HIGH);
+  DestroyAndReopen(opts);
+  SyncPoint::GetInstance()->ClearAllCallBacks();
+  SyncPoint::GetInstance()->DisableProcessing();
+  std::atomic_int flush_write_table_count = 0;
+  SyncPoint::GetInstance()->SetCallBack(
+      "FlushJob::WriteLevel0Table:s", [&](void* s_ptr) {
+        int c = flush_write_table_count.fetch_add(1);
+        if (c == 0) {
+          Status* s = (Status*)(s_ptr);
+          IOStatus io_error = IOStatus::IOError("injected foobar");
+          io_error.SetRetryable(true);
+          *s = io_error;
+        }
+      });
+
+  SyncPoint::GetInstance()->EnableProcessing();
+  SyncPoint::GetInstance()->LoadDependency(
+      {{"Let error recovery start",
+        "RecoverFromRetryableBGIOError:BeforeStart"},
+       {"RecoverFromRetryableBGIOError:RecoverSuccess",
+        "Wait for error recover"}});
+
+  ASSERT_OK(Put(Key(1), "val1"));
+  // trigger bg flush0 for mem0
+  ASSERT_OK(Put(Key(2), "val2"));
+  // Not checking status since this wait can finish before flush starts.
+  dbfull()->TEST_WaitForFlushMemTable().PermitUncheckedError();
+
+  // trigger bg flush1 for mem1, should see bg error and abort
+  // before picking a memtable to flush
+  ASSERT_OK(Put(Key(3), "val3"));
+  ASSERT_NOK(dbfull()->TEST_WaitForFlushMemTable());
+  ASSERT_EQ(0, NumTableFilesAtLevel(0));
+
+  TEST_SYNC_POINT("Let error recovery start");
+  TEST_SYNC_POINT("Wait for error recover");
+  // Recovery flush writes 2 memtables together into 1 file.
+  ASSERT_EQ(1, NumTableFilesAtLevel(0));
+  // 1 for flush 0 and 1 for recovery flush
+  ASSERT_EQ(2, flush_write_table_count);
+  SyncPoint::GetInstance()->ClearAllCallBacks();
+  SyncPoint::GetInstance()->DisableProcessing();
+}
+
+TEST_F(DBFlushTest, DBStuckAfterAtomicFlushError) {
+  // Test for a bug with atomic flush where DB can become stuck
+  // after a flush error. A repro timeline:
+  //
+  // Start Flush0 for mem0
+  // Start Flush1 for mem1
+  // Now Flush1 will wait for Flush0 to install mem0
+  // Flush0 finishes with retryable IOError, rollbacks mem0
+  // Resume starts and waits for background job to finish, i.e., Flush1
+  // Fill memtable again, trigger Flush2 for mem0
+  // Flush2 will get error status, and not rollback mem0, see code in
+  // https://github.com/facebook/rocksdb/blob/b927ba5936216861c2c35ab68f50ba4a78e65747/db/db_impl/db_impl_compaction_flush.cc#L725
+  //
+  // DB is stuck since mem0 can never be picked now
+  //
+  // The fix is to rollback mem0 in Flush2, and let Flush1 also abort upon
+  // background error besides waiting for older memtables to be installed.
+  // The recovery flush in this case should pick up all memtables
+  // and write them to a single L0 file.
+  Options opts = CurrentOptions();
+  opts.atomic_flush = true;
+  opts.memtable_factory.reset(test::NewSpecialSkipListFactory(1));
+  opts.max_write_buffer_number = 64;
+  opts.max_background_flushes = 4;
+  env_->SetBackgroundThreads(4, Env::HIGH);
+  DestroyAndReopen(opts);
+
+  std::atomic_int flush_count = 0;
+  SyncPoint::GetInstance()->ClearAllCallBacks();
+  SyncPoint::GetInstance()->DisableProcessing();
+  SyncPoint::GetInstance()->SetCallBack(
+      "FlushJob::WriteLevel0Table:s", [&](void* s_ptr) {
+        int c = flush_count.fetch_add(1);
+        if (c == 0) {
+          Status* s = (Status*)(s_ptr);
+          IOStatus io_error = IOStatus::IOError("injected foobar");
+          io_error.SetRetryable(true);
+          *s = io_error;
+          TEST_SYNC_POINT("Let flush for mem1 start");
+          // Wait for Flush1 to start waiting to install flush result
+          TEST_SYNC_POINT("Wait for flush for mem1");
+        }
+      });
+  SyncPoint::GetInstance()->LoadDependency(
+      {{"Let flush for mem1 start", "Flush for mem1"},
+       {"DBImpl::AtomicFlushMemTablesToOutputFiles:WaitCV",
+        "Wait for flush for mem1"},
+       {"RecoverFromRetryableBGIOError:BeforeStart",
+        "Wait for resume to start"},
+       {"Recovery should continue here",
+        "RecoverFromRetryableBGIOError:BeforeStart2"},
+       {"RecoverFromRetryableBGIOError:RecoverSuccess",
+        "Wait for error recover"}});
+  SyncPoint::GetInstance()->EnableProcessing();
+  ASSERT_OK(Put(Key(1), "val1"));
+  // trigger Flush0 for mem0
+  ASSERT_OK(Put(Key(2), "val2"));
+
+  // trigger Flush1 for mem1
+  TEST_SYNC_POINT("Flush for mem1");
+  ASSERT_OK(Put(Key(3), "val3"));
+
+  // Wait until resume started to schedule another flush
+  TEST_SYNC_POINT("Wait for resume to start");
+  // This flush should not be scheduled due to bg error
+  ASSERT_OK(Put(Key(4), "val4"));
+
+  // TEST_WaitForBackgroundWork() returns background error
+  // after all background work is done.
+  ASSERT_NOK(dbfull()->TEST_WaitForBackgroundWork());
+  // Flush should abort and not writing any table
+  ASSERT_EQ(0, NumTableFilesAtLevel(0));
+
+  // Wait until this flush is done.
+  TEST_SYNC_POINT("Recovery should continue here");
+  TEST_SYNC_POINT("Wait for error recover");
+  // error recovery can schedule new flushes, but should not
+  // encounter error
+  ASSERT_OK(dbfull()->TEST_WaitForBackgroundWork());
+  ASSERT_EQ(1, NumTableFilesAtLevel(0));
+}
 }  // namespace ROCKSDB_NAMESPACE
 
 int main(int argc, char** argv) {

db/db_impl/db_impl.cc

@@ -471,6 +471,28 @@ Status DBImpl::ResumeImpl(DBRecoverContext context) {
   if (shutdown_initiated_) {
     s = Status::ShutdownInProgress();
   }
+  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.
+    FlushOptions flush_opts;
+    flush_opts.allow_write_stall = false;
+    flush_opts.wait = false;
+    Status status = FlushAllColumnFamilies(
+        flush_opts, FlushReason::kCatchUpAfterErrorRecovery);
+    if (!status.ok()) {
+      // FlushAllColumnFamilies internally should take care of setting
+      // background error if needed.
+      ROCKS_LOG_INFO(immutable_db_options_.info_log,
+                     "The catch up flush after successful recovery failed [%s]",
+                     s.ToString().c_str());
+    }
+    // FlushAllColumnFamilies releases and re-acquires mutex.
+    if (shutdown_initiated_) {
+      s = Status::ShutdownInProgress();
+    }
+  }
+
   if (s.ok()) {
     for (auto cfd : *versions_->GetColumnFamilySet()) {
       SchedulePendingCompaction(cfd);
@@ -6043,7 +6065,8 @@ Status DBImpl::VerifyFullFileChecksum(const std::string& file_checksum_expected,
       fs_.get(), fname, immutable_db_options_.file_checksum_gen_factory.get(),
       func_name_expected, &file_checksum, &func_name,
       read_options.readahead_size, immutable_db_options_.allow_mmap_reads,
-      io_tracer_, immutable_db_options_.rate_limiter.get(), read_options);
+      io_tracer_, immutable_db_options_.rate_limiter.get(), read_options,
+      immutable_db_options_.stats, immutable_db_options_.clock);
   if (s.ok()) {
     assert(func_name_expected == func_name);
     if (file_checksum != file_checksum_expected) {

db/db_impl/db_impl_compaction_flush.cc

@@ -283,6 +283,24 @@ Status DBImpl::FlushMemTableToOutputFile(
   // If the log sync failed, we do not need to pick memtable. Otherwise,
   // num_flush_not_started_ needs to be rollback.
   TEST_SYNC_POINT("DBImpl::FlushMemTableToOutputFile:BeforePickMemtables");
+  // Exit a flush due to bg error should not set bg error again.
+  bool skip_set_bg_error = false;
+  if (s.ok() && !error_handler_.GetBGError().ok() &&
+      error_handler_.IsBGWorkStopped() &&
+      flush_reason != FlushReason::kErrorRecovery &&
+      flush_reason != FlushReason::kErrorRecoveryRetryFlush) {
+    // Error recovery in progress, should not pick memtable which excludes
+    // them from being picked up by recovery flush.
+    // This ensures that when bg error is set, no new flush can pick
+    // memtables.
+    skip_set_bg_error = true;
+    s = error_handler_.GetBGError();
+    assert(!s.ok());
+    ROCKS_LOG_BUFFER(log_buffer,
+                     "[JOB %d] Skip flush due to background error %s",
+                     job_context->job_id, s.ToString().c_str());
+  }
+
   if (s.ok()) {
     flush_job.PickMemTable();
     need_cancel = true;
@@ -303,7 +321,8 @@ Status DBImpl::FlushMemTableToOutputFile(
   // is unlocked by the current thread.
   if (s.ok()) {
     s = flush_job.Run(&logs_with_prep_tracker_, &file_meta,
-                      &switched_to_mempurge);
+                      &switched_to_mempurge, &skip_set_bg_error,
+                      &error_handler_);
     need_cancel = false;
   }
 
@@ -344,7 +363,8 @@ Status DBImpl::FlushMemTableToOutputFile(
     }
   }
 
-  if (!s.ok() && !s.IsShutdownInProgress() && !s.IsColumnFamilyDropped()) {
+  if (!s.ok() && !s.IsShutdownInProgress() && !s.IsColumnFamilyDropped() &&
+      !skip_set_bg_error) {
     if (log_io_s.ok()) {
       // Error while writing to MANIFEST.
       // In fact, versions_->io_status() can also be the result of renaming
@@ -556,6 +576,21 @@ Status DBImpl::AtomicFlushMemTablesToOutputFiles(
     pick_status.push_back(false);
   }
 
+  bool flush_for_recovery =
+      bg_flush_args[0].flush_reason_ == FlushReason::kErrorRecovery ||
+      bg_flush_args[0].flush_reason_ == FlushReason::kErrorRecoveryRetryFlush;
+  bool skip_set_bg_error = false;
+
+  if (s.ok() && !error_handler_.GetBGError().ok() &&
+      error_handler_.IsBGWorkStopped() && !flush_for_recovery) {
+    s = error_handler_.GetBGError();
+    skip_set_bg_error = true;
+    assert(!s.ok());
+    ROCKS_LOG_BUFFER(log_buffer,
+                     "[JOB %d] Skip flush due to background error %s",
+                     job_context->job_id, s.ToString().c_str());
+  }
+
   if (s.ok()) {
     for (int i = 0; i != num_cfs; ++i) {
       jobs[i]->PickMemTable();
@@ -620,7 +655,10 @@ Status DBImpl::AtomicFlushMemTablesToOutputFiles(
         }
       }
     }
-  } else {
+  } else if (!skip_set_bg_error) {
+    // When `skip_set_bg_error` is true, no memtable is picked so
+    // there is no need to call Cancel() or RollbackMemtableFlush().
+    //
     // Need to undo atomic flush if something went wrong, i.e. s is not OK and
     // it is not because of CF drop.
     // Have to cancel the flush jobs that have NOT executed because we need to
@@ -633,8 +671,8 @@ Status DBImpl::AtomicFlushMemTablesToOutputFiles(
     for (int i = 0; i != num_cfs; ++i) {
       if (exec_status[i].second.ok() && exec_status[i].first) {
         auto& mems = jobs[i]->GetMemTables();
-        cfds[i]->imm()->RollbackMemtableFlush(mems,
-                                              file_meta[i].fd.GetNumber());
+        cfds[i]->imm()->RollbackMemtableFlush(
+            mems, /*rollback_succeeding_memtables=*/false);
       }
     }
   }
@@ -676,10 +714,7 @@ Status DBImpl::AtomicFlushMemTablesToOutputFiles(
     };
 
     bool resuming_from_bg_err =
-        error_handler_.IsDBStopped() ||
-        (bg_flush_args[0].flush_reason_ == FlushReason::kErrorRecovery ||
-         bg_flush_args[0].flush_reason_ ==
-             FlushReason::kErrorRecoveryRetryFlush);
+        error_handler_.IsDBStopped() || flush_for_recovery;
     while ((!resuming_from_bg_err || error_handler_.GetRecoveryError().ok())) {
       std::pair<Status, bool> res = wait_to_install_func();
 
@@ -690,15 +725,27 @@ Status DBImpl::AtomicFlushMemTablesToOutputFiles(
         s = res.first;
         break;
       } else if (!res.second) {
+        // we are the oldest immutable memtable
+        break;
+      }
+      // We are not the oldest immutable memtable
+      TEST_SYNC_POINT_CALLBACK(
+          "DBImpl::AtomicFlushMemTablesToOutputFiles:WaitCV", &res);
+      //
+      // If bg work is stopped, recovery thread first calls
+      // WaitForBackgroundWork() before proceeding to flush for recovery. This
+      // flush can block WaitForBackgroundWork() while waiting for recovery
+      // flush to install result. To avoid this deadlock, we should abort here
+      // if there is background error.
+      if (!flush_for_recovery && error_handler_.IsBGWorkStopped() &&
+          !error_handler_.GetBGError().ok()) {
+        s = error_handler_.GetBGError();
+        assert(!s.ok());
         break;
       }
       atomic_flush_install_cv_.Wait();
 
-      resuming_from_bg_err =
-          error_handler_.IsDBStopped() ||
-          (bg_flush_args[0].flush_reason_ == FlushReason::kErrorRecovery ||
-           bg_flush_args[0].flush_reason_ ==
-               FlushReason::kErrorRecoveryRetryFlush);
+      resuming_from_bg_err = error_handler_.IsDBStopped() || flush_for_recovery;
     }
 
     if (!resuming_from_bg_err) {
@@ -714,6 +761,17 @@ Status DBImpl::AtomicFlushMemTablesToOutputFiles(
       // installation.
       s = error_handler_.GetRecoveryError();
     }
+    // Since we are not installing these memtables, need to rollback
+    // to allow future flush job to pick up these memtables.
+    if (!s.ok()) {
+      for (int i = 0; i != num_cfs; ++i) {
+        assert(exec_status[i].first);
+        assert(exec_status[i].second.ok());
+        auto& mems = jobs[i]->GetMemTables();
+        cfds[i]->imm()->RollbackMemtableFlush(
+            mems, /*rollback_succeeding_memtables=*/false);
+      }
+    }
   }
 
   if (s.ok()) {
@@ -817,7 +875,7 @@ Status DBImpl::AtomicFlushMemTablesToOutputFiles(
 
   // Need to undo atomic flush if something went wrong, i.e. s is not OK and
   // it is not because of CF drop.
-  if (!s.ok() && !s.IsColumnFamilyDropped()) {
+  if (!s.ok() && !s.IsColumnFamilyDropped() && !skip_set_bg_error) {
     if (log_io_s.ok()) {
       // Error while writing to MANIFEST.
       // In fact, versions_->io_status() can also be the result of renaming
@@ -2223,9 +2281,13 @@ Status DBImpl::FlushMemTable(ColumnFamilyData* cfd,
     WaitForPendingWrites();
 
     if (flush_reason != FlushReason::kErrorRecoveryRetryFlush &&
+        flush_reason != FlushReason::kCatchUpAfterErrorRecovery &&
         (!cfd->mem()->IsEmpty() || !cached_recoverable_state_empty_.load())) {
       // Note that, when flush reason is kErrorRecoveryRetryFlush, during the
       // auto retry resume, we want to avoid creating new small memtables.
+      // If flush reason is kCatchUpAfterErrorRecovery, we try to flush any new
+      // memtable that filled up during recovery, and we also want to avoid
+      // switching memtable to create small memtables.
       // Therefore, SwitchMemtable will not be called. Also, since ResumeImpl
       // will iterate through all the CFs and call FlushMemtable during auto
       // retry resume, it is possible that in some CFs,
@@ -2416,7 +2478,8 @@ Status DBImpl::AtomicFlushMemTables(
 
     for (auto cfd : cfds) {
       if ((cfd->mem()->IsEmpty() && cached_recoverable_state_empty_.load()) ||
-          flush_reason == FlushReason::kErrorRecoveryRetryFlush) {
+          flush_reason == FlushReason::kErrorRecoveryRetryFlush ||
+          flush_reason == FlushReason::kCatchUpAfterErrorRecovery) {
         continue;
       }
       cfd->Ref();
@@ -2684,6 +2747,11 @@ void DBImpl::MaybeScheduleFlushOrCompaction() {
     // There has been a hard error and this call is not part of the recovery
     // sequence. Bail out here so we don't get into an endless loop of
     // scheduling BG work which will again call this function
+    //
+    // Note that a non-recovery flush can still be scheduled if
+    // error_handler_.IsRecoveryInProgress() returns true. We rely on
+    // BackgroundCallFlush() to check flush reason and drop non-recovery
+    // flushes.
     return;
   } else if (shutting_down_.load(std::memory_order_acquire)) {
     // DB is being deleted; no more background compactions
@@ -3013,6 +3081,24 @@ Status DBImpl::BackgroundFlush(bool* made_progress, JobContext* job_context,
     // This cfd is already referenced
     FlushRequest flush_req = PopFirstFromFlushQueue();
     FlushReason flush_reason = flush_req.flush_reason;
+    if (!error_handler_.GetBGError().ok() && error_handler_.IsBGWorkStopped() &&
+        flush_reason != FlushReason::kErrorRecovery &&
+        flush_reason != FlushReason::kErrorRecoveryRetryFlush) {
+      // Stop non-recovery flush when bg work is stopped
+      // Note that we drop the flush request here.
+      // Recovery thread should schedule further flushes after bg error
+      // is cleared.
+      status = error_handler_.GetBGError();
+      assert(!status.ok());
+      ROCKS_LOG_BUFFER(log_buffer,
+                       "[JOB %d] Abort flush due to background error %s",
+                       job_context->job_id, status.ToString().c_str());
+      *reason = flush_reason;
+      for (auto item : flush_req.cfd_to_max_mem_id_to_persist) {
+        item.first->UnrefAndTryDelete();
+      }
+      return status;
+    }
     if (!immutable_db_options_.atomic_flush &&
         ShouldRescheduleFlushRequestToRetainUDT(flush_req)) {
       assert(flush_req.cfd_to_max_mem_id_to_persist.size() == 1);
@@ -3155,9 +3241,9 @@ void DBImpl::BackgroundCallFlush(Env::Priority thread_pri) {
       bg_cv_.SignalAll();  // In case a waiter can proceed despite the error
       mutex_.Unlock();
       ROCKS_LOG_ERROR(immutable_db_options_.info_log,
-                      "Waiting after background flush error: %s"
+                      "[JOB %d] Waiting after background flush error: %s"
                       "Accumulated background error counts: %" PRIu64,
-                      s.ToString().c_str(), error_cnt);
+                      job_context.job_id, s.ToString().c_str(), error_cnt);
       log_buffer.FlushBufferToLog();
       LogFlush(immutable_db_options_.info_log);
       immutable_db_options_.clock->SleepForMicroseconds(1000000);

db/db_impl/db_impl_open.cc

@@ -143,13 +143,6 @@ DBOptions SanitizeOptions(const std::string& dbname, const DBOptions& src,
     result.wal_dir = result.wal_dir.substr(0, result.wal_dir.size() - 1);
   }
 
-  if (result.compaction_readahead_size == 0) {
-    if (result.use_direct_reads) {
-      TEST_SYNC_POINT_CALLBACK("SanitizeOptions:direct_io", nullptr);
-    }
-    result.compaction_readahead_size = 1024 * 1024 * 2;
-  }
-
   // Force flush on DB open if 2PC is enabled, since with 2PC we have no
   // guarantee that consecutive log files have consecutive sequence id, which
   // make recovery complicated.

db/db_options_test.cc

@@ -1034,30 +1034,37 @@ TEST_F(DBOptionsTest, SetFIFOCompactionOptions) {
 }
 
 TEST_F(DBOptionsTest, CompactionReadaheadSizeChange) {
-  SpecialEnv env(env_);
-  Options options;
-  options.env = &env;
+  for (bool use_direct_reads : {true, false}) {
+    SpecialEnv env(env_);
+    Options options;
+    options.env = &env;
 
-  options.compaction_readahead_size = 0;
-  options.level0_file_num_compaction_trigger = 2;
-  const std::string kValue(1024, 'v');
-  Reopen(options);
+    options.use_direct_reads = use_direct_reads;
+    options.level0_file_num_compaction_trigger = 2;
+    const std::string kValue(1024, 'v');
+    Status s = TryReopen(options);
+    if (use_direct_reads && (s.IsNotSupported() || s.IsInvalidArgument())) {
+      continue;
+    } else {
+      ASSERT_OK(s);
+    }
 
-  ASSERT_EQ(1024 * 1024 * 2,
-            dbfull()->GetDBOptions().compaction_readahead_size);
-  ASSERT_OK(dbfull()->SetDBOptions({{"compaction_readahead_size", "256"}}));
-  ASSERT_EQ(256, dbfull()->GetDBOptions().compaction_readahead_size);
-  for (int i = 0; i < 1024; i++) {
-    ASSERT_OK(Put(Key(i), kValue));
+    ASSERT_EQ(1024 * 1024 * 2,
+              dbfull()->GetDBOptions().compaction_readahead_size);
+    ASSERT_OK(dbfull()->SetDBOptions({{"compaction_readahead_size", "256"}}));
+    ASSERT_EQ(256, dbfull()->GetDBOptions().compaction_readahead_size);
+    for (int i = 0; i < 1024; i++) {
+      ASSERT_OK(Put(Key(i), kValue));
+    }
+    ASSERT_OK(Flush());
+    for (int i = 0; i < 1024 * 2; i++) {
+      ASSERT_OK(Put(Key(i), kValue));
+    }
+    ASSERT_OK(Flush());
+    ASSERT_OK(dbfull()->TEST_WaitForCompact());
+    ASSERT_EQ(256, env_->compaction_readahead_size_);
+    Close();
   }
-  ASSERT_OK(Flush());
-  for (int i = 0; i < 1024 * 2; i++) {
-    ASSERT_OK(Put(Key(i), kValue));
-  }
-  ASSERT_OK(Flush());
-  ASSERT_OK(dbfull()->TEST_WaitForCompact());
-  ASSERT_EQ(256, env_->compaction_readahead_size_);
-  Close();
 }
 
 TEST_F(DBOptionsTest, FIFOTtlBackwardCompatible) {

db/db_test2.cc

@@ -6876,6 +6876,7 @@ TEST_F(DBTest2, LastLevelTemperatureUniversal) {
 TEST_F(DBTest2, LastLevelStatistics) {
   Options options = CurrentOptions();
   options.bottommost_temperature = Temperature::kWarm;
+  options.default_temperature = Temperature::kHot;
   options.level0_file_num_compaction_trigger = 2;
   options.level_compaction_dynamic_level_bytes = true;
   options.statistics = CreateDBStatistics();
@@ -6889,6 +6890,10 @@ TEST_F(DBTest2, LastLevelStatistics) {
 
   ASSERT_GT(options.statistics->getTickerCount(NON_LAST_LEVEL_READ_BYTES), 0);
   ASSERT_GT(options.statistics->getTickerCount(NON_LAST_LEVEL_READ_COUNT), 0);
+  ASSERT_EQ(options.statistics->getTickerCount(NON_LAST_LEVEL_READ_BYTES),
+            options.statistics->getTickerCount(HOT_FILE_READ_BYTES));
+  ASSERT_EQ(options.statistics->getTickerCount(NON_LAST_LEVEL_READ_COUNT),
+            options.statistics->getTickerCount(HOT_FILE_READ_COUNT));
   ASSERT_EQ(options.statistics->getTickerCount(LAST_LEVEL_READ_BYTES), 0);
   ASSERT_EQ(options.statistics->getTickerCount(LAST_LEVEL_READ_COUNT), 0);
 
@@ -6899,6 +6904,10 @@ TEST_F(DBTest2, LastLevelStatistics) {
   ASSERT_OK(dbfull()->TEST_WaitForCompact());
   ASSERT_EQ("bar", Get("bar"));
 
+  ASSERT_EQ(options.statistics->getTickerCount(NON_LAST_LEVEL_READ_BYTES),
+            options.statistics->getTickerCount(HOT_FILE_READ_BYTES));
+  ASSERT_EQ(options.statistics->getTickerCount(NON_LAST_LEVEL_READ_COUNT),
+            options.statistics->getTickerCount(HOT_FILE_READ_COUNT));
   ASSERT_EQ(options.statistics->getTickerCount(LAST_LEVEL_READ_BYTES),
             options.statistics->getTickerCount(WARM_FILE_READ_BYTES));
   ASSERT_EQ(options.statistics->getTickerCount(LAST_LEVEL_READ_COUNT),
@@ -6919,6 +6928,30 @@ TEST_F(DBTest2, LastLevelStatistics) {
             pre_bytes);
   ASSERT_GT(options.statistics->getTickerCount(NON_LAST_LEVEL_READ_COUNT),
             pre_count);
+  ASSERT_EQ(options.statistics->getTickerCount(NON_LAST_LEVEL_READ_BYTES),
+            options.statistics->getTickerCount(HOT_FILE_READ_BYTES));
+  ASSERT_EQ(options.statistics->getTickerCount(NON_LAST_LEVEL_READ_COUNT),
+            options.statistics->getTickerCount(HOT_FILE_READ_COUNT));
+  ASSERT_EQ(options.statistics->getTickerCount(LAST_LEVEL_READ_BYTES),
+            options.statistics->getTickerCount(WARM_FILE_READ_BYTES));
+  ASSERT_EQ(options.statistics->getTickerCount(LAST_LEVEL_READ_COUNT),
+            options.statistics->getTickerCount(WARM_FILE_READ_COUNT));
+
+  // Not a realistic setting to make last level kWarm and default temp kCold.
+  // This is just for testing default temp can be reset on reopen while the
+  // last level temp is consistent across DB reopen because those file's temp
+  // are persisted in manifest.
+  options.default_temperature = Temperature::kCold;
+  ASSERT_OK(options.statistics->Reset());
+  Reopen(options);
+  ASSERT_EQ("bar", Get("bar"));
+
+  ASSERT_EQ(0, options.statistics->getTickerCount(HOT_FILE_READ_BYTES));
+
+  ASSERT_EQ(options.statistics->getTickerCount(NON_LAST_LEVEL_READ_BYTES),
+            options.statistics->getTickerCount(COLD_FILE_READ_BYTES));
+  ASSERT_EQ(options.statistics->getTickerCount(NON_LAST_LEVEL_READ_COUNT),
+            options.statistics->getTickerCount(COLD_FILE_READ_COUNT));
   ASSERT_EQ(options.statistics->getTickerCount(LAST_LEVEL_READ_BYTES),
             options.statistics->getTickerCount(WARM_FILE_READ_BYTES));
   ASSERT_EQ(options.statistics->getTickerCount(LAST_LEVEL_READ_COUNT),

db/error_handler.cc

@@ -653,6 +653,7 @@ const Status& ErrorHandler::StartRecoverFromRetryableBGIOError(
   }
 
   recovery_in_prog_ = true;
+  TEST_SYNC_POINT("StartRecoverFromRetryableBGIOError::in_progress");
   recovery_thread_.reset(
       new port::Thread(&ErrorHandler::RecoverFromRetryableBGIOError, this));
 
@@ -667,6 +668,7 @@ const Status& ErrorHandler::StartRecoverFromRetryableBGIOError(
 // mutex is released.
 void ErrorHandler::RecoverFromRetryableBGIOError() {
   TEST_SYNC_POINT("RecoverFromRetryableBGIOError:BeforeStart");
+  TEST_SYNC_POINT("RecoverFromRetryableBGIOError:BeforeStart2");
   InstrumentedMutexLock l(db_mutex_);
   if (end_recovery_) {
     EventHelpers::NotifyOnErrorRecoveryEnd(db_options_.listeners, bg_error_,
@@ -675,6 +677,7 @@ void ErrorHandler::RecoverFromRetryableBGIOError() {
     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;

db/error_handler.h

@@ -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 {

db/event_helpers.cc

@@ -240,6 +240,8 @@ void EventHelpers::NotifyOnErrorRecoveryEnd(
       info.new_bg_error.PermitUncheckedError();
     }
     db_mutex->Lock();
+  } else {
+    old_bg_error.PermitUncheckedError();
   }
 }
 

db/external_sst_file_ingestion_job.cc

@@ -226,7 +226,8 @@ Status ExternalSstFileIngestionJob::Prepare(
             &generated_checksum_func_name,
             ingestion_options_.verify_checksums_readahead_size,
             db_options_.allow_mmap_reads, io_tracer_,
-            db_options_.rate_limiter.get(), ro);
+            db_options_.rate_limiter.get(), ro, db_options_.stats,
+            db_options_.clock);
         if (!io_s.ok()) {
           status = io_s;
           ROCKS_LOG_WARN(db_options_.info_log,
@@ -1067,7 +1068,7 @@ IOStatus ExternalSstFileIngestionJob::GenerateChecksumForIngestedFile(
       &file_checksum, &file_checksum_func_name,
       ingestion_options_.verify_checksums_readahead_size,
       db_options_.allow_mmap_reads, io_tracer_, db_options_.rate_limiter.get(),
-      ro);
+      ro, db_options_.stats, db_options_.clock);
   if (!io_s.ok()) {
     return io_s;
   }

db/flush_job.cc

@@ -79,6 +79,8 @@ const char* GetFlushReasonString(FlushReason flush_reason) {
       return "Error Recovery Retry Flush";
     case FlushReason::kWalFull:
       return "WAL Full";
+    case FlushReason::kCatchUpAfterErrorRecovery:
+      return "Catch Up After Error Recovery";
     default:
       return "Invalid";
   }
@@ -215,7 +217,8 @@ void FlushJob::PickMemTable() {
 }
 
 Status FlushJob::Run(LogsWithPrepTracker* prep_tracker, FileMetaData* file_meta,
-                     bool* switched_to_mempurge) {
+                     bool* switched_to_mempurge, bool* skipped_since_bg_error,
+                     ErrorHandler* error_handler) {
   TEST_SYNC_POINT("FlushJob::Start");
   db_mutex_->AssertHeld();
   assert(pick_memtable_called);
@@ -303,17 +306,32 @@ Status FlushJob::Run(LogsWithPrepTracker* prep_tracker, FileMetaData* file_meta,
   }
 
   if (!s.ok()) {
-    cfd_->imm()->RollbackMemtableFlush(mems_, meta_.fd.GetNumber());
+    cfd_->imm()->RollbackMemtableFlush(
+        mems_, /*rollback_succeeding_memtables=*/!db_options_.atomic_flush);
   } else if (write_manifest_) {
-    TEST_SYNC_POINT("FlushJob::InstallResults");
-    // Replace immutable memtable with the generated Table
-    s = cfd_->imm()->TryInstallMemtableFlushResults(
-        cfd_, mutable_cf_options_, mems_, prep_tracker, versions_, db_mutex_,
-        meta_.fd.GetNumber(), &job_context_->memtables_to_free, db_directory_,
-        log_buffer_, &committed_flush_jobs_info_,
-        !(mempurge_s.ok()) /* write_edit : true if no mempurge happened (or if aborted),
+    assert(!db_options_.atomic_flush);
+    if (!db_options_.atomic_flush &&
+        flush_reason_ != FlushReason::kErrorRecovery &&
+        flush_reason_ != FlushReason::kErrorRecoveryRetryFlush &&
+        error_handler && !error_handler->GetBGError().ok() &&
+        error_handler->IsBGWorkStopped()) {
+      cfd_->imm()->RollbackMemtableFlush(
+          mems_, /*rollback_succeeding_memtables=*/!db_options_.atomic_flush);
+      s = error_handler->GetBGError();
+      if (skipped_since_bg_error) {
+        *skipped_since_bg_error = true;
+      }
+    } else {
+      TEST_SYNC_POINT("FlushJob::InstallResults");
+      // Replace immutable memtable with the generated Table
+      s = cfd_->imm()->TryInstallMemtableFlushResults(
+              cfd_, mutable_cf_options_, mems_, prep_tracker, versions_, db_mutex_,
+              meta_.fd.GetNumber(), &job_context_->memtables_to_free, db_directory_,
+              log_buffer_, &committed_flush_jobs_info_,
+              !(mempurge_s.ok()) /* write_edit : true if no mempurge happened (or if aborted),
                               but 'false' if mempurge successful: no new min log number
                               or new level 0 file path to write to manifest. */);
+    }
   }
 
   if (s.ok() && file_meta != nullptr) {
@@ -965,6 +983,7 @@ Status FlushJob::WriteLevel0Table() {
                      &table_properties_, write_hint, full_history_ts_low,
                      blob_callback_, base_, &num_input_entries,
                      &memtable_payload_bytes, &memtable_garbage_bytes);
+      TEST_SYNC_POINT_CALLBACK("FlushJob::WriteLevel0Table:s", &s);
       // TODO: Cleanup io_status in BuildTable and table builders
       assert(!s.ok() || io_s.ok());
       io_s.PermitUncheckedError();

db/flush_job.h

@@ -83,9 +83,14 @@ class FlushJob {
   // Require db_mutex held.
   // Once PickMemTable() is called, either Run() or Cancel() has to be called.
   void PickMemTable();
+  // @param skip_since_bg_error If not nullptr and if atomic_flush=false,
+  // then it is set to true if flush installation is skipped and memtable
+  // is rolled back due to existing background error.
   Status Run(LogsWithPrepTracker* prep_tracker = nullptr,
              FileMetaData* file_meta = nullptr,
-             bool* switched_to_mempurge = nullptr);
+             bool* switched_to_mempurge = nullptr,
+             bool* skipped_since_bg_error = nullptr,
+             ErrorHandler* error_handler = nullptr);
   void Cancel();
   const autovector<MemTable*>& GetMemTables() const { return mems_; }
 

db/memtable_list.cc

@@ -434,23 +434,57 @@ void MemTableList::PickMemtablesToFlush(uint64_t max_memtable_id,
 }
 
 void MemTableList::RollbackMemtableFlush(const autovector<MemTable*>& mems,
-                                         uint64_t /*file_number*/) {
+                                         bool rollback_succeeding_memtables) {
+  TEST_SYNC_POINT("RollbackMemtableFlush");
   AutoThreadOperationStageUpdater stage_updater(
       ThreadStatus::STAGE_MEMTABLE_ROLLBACK);
-  assert(!mems.empty());
-
-  // If the flush was not successful, then just reset state.
-  // Maybe a succeeding attempt to flush will be successful.
+#ifndef NDEBUG
   for (MemTable* m : mems) {
     assert(m->flush_in_progress_);
     assert(m->file_number_ == 0);
-
-    m->flush_in_progress_ = false;
-    m->flush_completed_ = false;
-    m->edit_.Clear();
-    num_flush_not_started_++;
   }
-  imm_flush_needed.store(true, std::memory_order_release);
+#endif
+
+  if (rollback_succeeding_memtables && !mems.empty()) {
+    std::list<MemTable*>& memlist = current_->memlist_;
+    auto it = memlist.rbegin();
+    for (; *it != mems[0] && it != memlist.rend(); ++it) {
+    }
+    // mems should be in memlist
+    assert(*it == mems[0]);
+    if (*it == mems[0]) {
+      ++it;
+    }
+    while (it != memlist.rend()) {
+      MemTable* m = *it;
+      // Only rollback complete, not in-progress,
+      // in_progress can be flushes that are still writing SSTs
+      if (m->flush_completed_) {
+        m->flush_in_progress_ = false;
+        m->flush_completed_ = false;
+        m->edit_.Clear();
+        m->file_number_ = 0;
+        num_flush_not_started_++;
+        ++it;
+      } else {
+        break;
+      }
+    }
+  }
+
+  for (MemTable* m : mems) {
+    if (m->flush_in_progress_) {
+      assert(m->file_number_ == 0);
+      m->file_number_ = 0;
+      m->flush_in_progress_ = false;
+      m->flush_completed_ = false;
+      m->edit_.Clear();
+      num_flush_not_started_++;
+    }
+  }
+  if (!mems.empty()) {
+    imm_flush_needed.store(true, std::memory_order_release);
+  }
 }
 
 // Try record a successful flush in the manifest file. It might just return

db/memtable_list.h

@@ -271,8 +271,20 @@ class MemTableList {
 
   // Reset status of the given memtable list back to pending state so that
   // they can get picked up again on the next round of flush.
+  //
+  // @param rollback_succeeding_memtables If true, will rollback adjacent
+  // younger memtables whose flush is completed. Specifically, suppose the
+  // current immutable memtables are M_0,M_1...M_N ordered from youngest to
+  // oldest. Suppose that the youngest memtable in `mems` is M_K. We will try to
+  // rollback M_K-1, M_K-2... until the first memtable whose flush is
+  // not completed. These are the memtables that would have been installed
+  // by this flush job if it were to succeed. This flag is currently used
+  // by non atomic_flush rollback.
+  // Note that we also do rollback in `write_manifest_cb` by calling
+  // `RemoveMemTablesOrRestoreFlags()`. There we rollback the entire batch so
+  // it is similar to what we do here with rollback_succeeding_memtables=true.
   void RollbackMemtableFlush(const autovector<MemTable*>& mems,
-                             uint64_t file_number);
+                             bool rollback_succeeding_memtables);
 
   // Try commit a successful flush in the manifest file. It might just return
   // Status::OK letting a concurrent flush to do the actual the recording.

db/memtable_list_test.cc

@@ -682,7 +682,7 @@ TEST_F(MemTableListTest, FlushPendingTest) {
   ASSERT_FALSE(list.imm_flush_needed.load(std::memory_order_acquire));
 
   // Revert flush
-  list.RollbackMemtableFlush(to_flush, 0);
+  list.RollbackMemtableFlush(to_flush, false);
   ASSERT_FALSE(list.IsFlushPending());
   ASSERT_TRUE(list.imm_flush_needed.load(std::memory_order_acquire));
   to_flush.clear();
@@ -732,7 +732,7 @@ TEST_F(MemTableListTest, FlushPendingTest) {
   ASSERT_FALSE(list.imm_flush_needed.load(std::memory_order_acquire));
 
   // Rollback first pick of tables
-  list.RollbackMemtableFlush(to_flush, 0);
+  list.RollbackMemtableFlush(to_flush, false);
   ASSERT_TRUE(list.IsFlushPending());
   ASSERT_TRUE(list.imm_flush_needed.load(std::memory_order_acquire));
   to_flush.clear();

db/table_cache.cc

@@ -129,6 +129,10 @@ Status TableCache::GetTableReader(
     if (!sequential_mode && ioptions_.advise_random_on_open) {
       file->Hint(FSRandomAccessFile::kRandom);
     }
+    if (ioptions_.default_temperature != Temperature::kUnknown &&
+        file_temperature == Temperature::kUnknown) {
+      file_temperature = ioptions_.default_temperature;
+    }
     StopWatch sw(ioptions_.clock, ioptions_.stats, TABLE_OPEN_IO_MICROS);
     std::unique_ptr<RandomAccessFileReader> file_reader(
         new RandomAccessFileReader(std::move(file), fname, ioptions_.clock,

db/version_builder.cc

@@ -1323,11 +1323,6 @@ class VersionBuilder::Rep {
 
         auto* file_meta = files_meta[file_idx].first;
         int level = files_meta[file_idx].second;
-        Temperature file_temperature = file_meta->temperature;
-        if (ioptions_->default_temperature != Temperature::kUnknown &&
-            file_temperature == Temperature::kUnknown) {
-          file_temperature = ioptions_->default_temperature;
-        }
         TableCache::TypedHandle* handle = nullptr;
         statuses[file_idx] = table_cache_->FindTable(
             read_options, file_options_,
@@ -1335,7 +1330,7 @@ class VersionBuilder::Rep {
             block_protection_bytes_per_key, prefix_extractor, false /*no_io */,
             internal_stats->GetFileReadHist(level), false, level,
             prefetch_index_and_filter_in_cache, max_file_size_for_l0_meta_pin,
-            file_temperature);
+            file_meta->temperature);
         if (handle != nullptr) {
           file_meta->table_reader_handle = handle;
           // Load table_reader

file/file_prefetch_buffer.cc

@@ -545,7 +545,9 @@ Status FilePrefetchBuffer::PrefetchAsyncInternal(const IOOptions& opts,
     assert(roundup_len1 >= chunk_len1);
     read_len1 = static_cast<size_t>(roundup_len1 - chunk_len1);
   }
-  {
+
+  // Prefetch in second buffer only if readahead_size_ > 0.
+  if (readahead_size_ > 0) {
     // offset and size alignment for second buffer for asynchronous
     // prefetching
     uint64_t rounddown_start2 = roundup_end1;
@@ -733,7 +735,9 @@ bool FilePrefetchBuffer::TryReadFromCacheAsyncUntracked(
       (bufs_[curr_].async_read_in_progress_ ||
        offset + n >
            bufs_[curr_].offset_ + bufs_[curr_].buffer_.CurrentSize())) {
-    if (readahead_size_ > 0) {
+    // In case readahead_size is trimmed (=0), we still want to poll the data
+    // submitted with explicit_prefetch_submitted_=true.
+    if (readahead_size_ > 0 || explicit_prefetch_submitted_) {
       Status s;
       assert(reader != nullptr);
       assert(max_readahead_size_ >= readahead_size_);
@@ -825,14 +829,12 @@ Status FilePrefetchBuffer::PrefetchAsync(const IOOptions& opts,
   num_file_reads_ = 0;
   explicit_prefetch_submitted_ = false;
   bool is_eligible_for_prefetching = false;
+
+  UpdateReadAheadSizeForUpperBound(offset, n);
   if (readahead_size_ > 0 &&
       (!implicit_auto_readahead_ ||
        num_file_reads_ >= num_file_reads_for_auto_readahead_)) {
-    UpdateReadAheadSizeForUpperBound(offset, n);
-    // After trim, readahead size can be 0.
-    if (readahead_size_ > 0) {
       is_eligible_for_prefetching = true;
-    }
   }
 
   // 1. Cancel any pending async read to make code simpler as buffers can be out
@@ -894,18 +896,24 @@ Status FilePrefetchBuffer::PrefetchAsync(const IOOptions& opts,
   //   - prefetch_size on second.
   // Calculate length and offsets for reading.
   if (!DoesBufferContainData(curr_)) {
+    uint64_t roundup_len1;
     // Prefetch full data + prefetch_size in curr_.
-    rounddown_start1 = Rounddown(offset_to_read, alignment);
-    roundup_end1 = Roundup(offset_to_read + n + prefetch_size, alignment);
-    uint64_t roundup_len1 = roundup_end1 - rounddown_start1;
-    assert(roundup_len1 >= alignment);
-    assert(roundup_len1 % alignment == 0);
-
+    if (is_eligible_for_prefetching || reader->use_direct_io()) {
+      rounddown_start1 = Rounddown(offset_to_read, alignment);
+      roundup_end1 = Roundup(offset_to_read + n + prefetch_size, alignment);
+      roundup_len1 = roundup_end1 - rounddown_start1;
+      assert(roundup_len1 >= alignment);
+      assert(roundup_len1 % alignment == 0);
+    } else {
+      rounddown_start1 = offset_to_read;
+      roundup_end1 = offset_to_read + n;
+      roundup_len1 = roundup_end1 - rounddown_start1;
+    }
     CalculateOffsetAndLen(alignment, rounddown_start1, roundup_len1, curr_,
                           false, chunk_len1);
     assert(chunk_len1 == 0);
     assert(roundup_len1 >= chunk_len1);
-    read_len1 = static_cast<size_t>(roundup_len1 - chunk_len1);
+    read_len1 = static_cast<size_t>(roundup_len1);
     bufs_[curr_].offset_ = rounddown_start1;
   }
 

file/file_prefetch_buffer.h

@@ -389,6 +389,12 @@ class FilePrefetchBuffer {
          bufs_[second].offset_)) {
       return false;
     }
+
+    // Readahead size can be 0 because of trimming.
+    if (readahead_size_ == 0) {
+      return false;
+    }
+
     bufs_[second].buffer_.Clear();
     return true;
   }
@@ -422,7 +428,8 @@ class FilePrefetchBuffer {
   void UpdateReadAheadSizeForUpperBound(uint64_t offset, size_t n) {
     // Adjust readhahead_size till upper_bound if upper_bound_offset_ is
     // set.
-    if (upper_bound_offset_ > 0 && upper_bound_offset_ > offset) {
+    if (readahead_size_ > 0 && upper_bound_offset_ > 0 &&
+        upper_bound_offset_ > offset) {
       if (upper_bound_offset_ < offset + n + readahead_size_) {
         readahead_size_ = (upper_bound_offset_ - offset) - n;
         RecordTick(stats_, READAHEAD_TRIMMED);

file/file_util.cc

@@ -13,6 +13,7 @@
 #include "file/sst_file_manager_impl.h"
 #include "file/writable_file_writer.h"
 #include "rocksdb/env.h"
+#include "rocksdb/statistics.h"
 
 namespace ROCKSDB_NAMESPACE {
 
@@ -137,7 +138,7 @@ IOStatus GenerateOneFileChecksum(
     std::string* file_checksum_func_name,
     size_t verify_checksums_readahead_size, bool /*allow_mmap_reads*/,
     std::shared_ptr<IOTracer>& io_tracer, RateLimiter* rate_limiter,
-    const ReadOptions& read_options) {
+    const ReadOptions& read_options, Statistics* stats, SystemClock* clock) {
   if (checksum_factory == nullptr) {
     return IOStatus::InvalidArgument("Checksum factory is invalid");
   }
@@ -186,8 +187,8 @@ IOStatus GenerateOneFileChecksum(
       return io_s;
     }
     reader.reset(new RandomAccessFileReader(
-        std::move(r_file), file_path, nullptr /*Env*/, io_tracer, nullptr,
-        Histograms::HISTOGRAM_ENUM_MAX, nullptr, rate_limiter));
+        std::move(r_file), file_path, clock, io_tracer, stats,
+        Histograms::SST_READ_MICROS, nullptr, rate_limiter));
   }
 
   // Found that 256 KB readahead size provides the best performance, based on

file/file_util.h

@@ -11,6 +11,7 @@
 #include "rocksdb/env.h"
 #include "rocksdb/file_system.h"
 #include "rocksdb/sst_file_writer.h"
+#include "rocksdb/statistics.h"
 #include "rocksdb/status.h"
 #include "rocksdb/system_clock.h"
 #include "rocksdb/types.h"
@@ -52,6 +53,7 @@ extern Status DeleteDBFile(const ImmutableDBOptions* db_options,
                            const std::string& path_to_sync, const bool force_bg,
                            const bool force_fg);
 
+// TODO(hx235): pass the whole DBOptions intead of its individual fields
 extern IOStatus GenerateOneFileChecksum(
     FileSystem* fs, const std::string& file_path,
     FileChecksumGenFactory* checksum_factory,
@@ -59,7 +61,7 @@ extern IOStatus GenerateOneFileChecksum(
     std::string* file_checksum_func_name,
     size_t verify_checksums_readahead_size, bool allow_mmap_reads,
     std::shared_ptr<IOTracer>& io_tracer, RateLimiter* rate_limiter,
-    const ReadOptions& read_options);
+    const ReadOptions& read_options, Statistics* stats, SystemClock* clock);
 
 inline IOStatus PrepareIOFromReadOptions(const ReadOptions& ro,
                                          SystemClock* clock, IOOptions& opts) {

file/prefetch_test.cc

@@ -2161,6 +2161,121 @@ TEST_P(PrefetchTest, IterReadAheadSizeWithUpperBound) {
   }
 }
 
+// This test checks if readahead_size is trimmed when upper_bound is reached
+// during Seek in async_io and it goes for polling without any extra
+// prefetching.
+TEST_P(PrefetchTest, IterReadAheadSizeWithUpperBoundSeekOnly) {
+  if (mem_env_ || encrypted_env_) {
+    ROCKSDB_GTEST_SKIP("Test requires non-mem or non-encrypted environment");
+    return;
+  }
+
+  // First param is if the mockFS support_prefetch or not
+  std::shared_ptr<MockFS> fs =
+      std::make_shared<MockFS>(FileSystem::Default(), false);
+
+  bool use_direct_io = false;
+  if (std::get<0>(GetParam())) {
+    use_direct_io = true;
+  }
+
+  std::unique_ptr<Env> env(new CompositeEnvWrapper(env_, fs));
+  Options options;
+  SetGenericOptions(env.get(), use_direct_io, options);
+  options.statistics = CreateDBStatistics();
+  BlockBasedTableOptions table_options;
+  SetBlockBasedTableOptions(table_options);
+  options.table_factory.reset(NewBlockBasedTableFactory(table_options));
+
+  Status s = TryReopen(options);
+  if (use_direct_io && (s.IsNotSupported() || s.IsInvalidArgument())) {
+    // If direct IO is not supported, skip the test
+    return;
+  } else {
+    ASSERT_OK(s);
+  }
+
+  Random rnd(309);
+  WriteBatch batch;
+
+  for (int i = 0; i < 26; i++) {
+    std::string key = "my_key_";
+
+    for (int j = 0; j < 10; j++) {
+      key += char('a' + i);
+      ASSERT_OK(batch.Put(key, rnd.RandomString(1000)));
+    }
+  }
+  ASSERT_OK(db_->Write(WriteOptions(), &batch));
+
+  std::string start_key = "my_key_a";
+
+  std::string end_key = "my_key_";
+  for (int j = 0; j < 10; j++) {
+    end_key += char('a' + 25);
+  }
+
+  Slice least(start_key.data(), start_key.size());
+  Slice greatest(end_key.data(), end_key.size());
+
+  ASSERT_OK(db_->CompactRange(CompactRangeOptions(), &least, &greatest));
+
+  s = TryReopen(options);
+  ASSERT_OK(s);
+
+  int buff_count_with_tuning = 0;
+
+  SyncPoint::GetInstance()->SetCallBack(
+      "FilePrefetchBuffer::PrefetchAsyncInternal:Start",
+      [&](void*) { buff_count_with_tuning++; });
+
+  bool read_async_called = false;
+  SyncPoint::GetInstance()->SetCallBack(
+      "UpdateResults::io_uring_result",
+      [&](void* /*arg*/) { read_async_called = true; });
+
+  SyncPoint::GetInstance()->EnableProcessing();
+
+  SyncPoint::GetInstance()->EnableProcessing();
+
+  ReadOptions ropts;
+  if (std::get<1>(GetParam())) {
+    ropts.readahead_size = 32768;
+  }
+  ropts.async_io = true;
+
+  Slice ub = Slice("my_key_aaa");
+  ropts.iterate_upper_bound = &ub;
+  Slice seek_key = Slice("my_key_aaa");
+
+  // With tuning readahead_size.
+  {
+    ASSERT_OK(options.statistics->Reset());
+    ropts.auto_readahead_size = true;
+
+    auto iter = std::unique_ptr<Iterator>(db_->NewIterator(ropts));
+
+    iter->Seek(seek_key);
+
+    ASSERT_OK(iter->status());
+
+    // Verify results.
+    uint64_t readhahead_trimmed =
+        options.statistics->getAndResetTickerCount(READAHEAD_TRIMMED);
+    // Readahead got trimmed.
+    if (read_async_called) {
+      ASSERT_GT(readhahead_trimmed, 0);
+      // Seek called PrefetchAsync to poll the data.
+      ASSERT_EQ(1, buff_count_with_tuning);
+    } else {
+      // async_io disabled.
+      ASSERT_GE(readhahead_trimmed, 0);
+      ASSERT_EQ(0, buff_count_with_tuning);
+    }
+  }
+  Close();
+}
+
 namespace {
 #ifdef GFLAGS
 const int kMaxArgCount = 100;
@@ -2759,6 +2874,89 @@ TEST_F(FilePrefetchBufferTest, SeekWithBlockCacheHit) {
       fpb.TryReadFromCacheAsync(io_opts, r.get(), 8192, 8192, &result, &s));
 }
 
+// Test to ensure when PrefetchAsync is called during seek, it doesn't do any
+// alignment or prefetch extra if readahead is not enabled during seek.
+TEST_F(FilePrefetchBufferTest, SeekWithoutAlignment) {
+  std::string fname = "seek-wwithout-alignment";
+  Random rand(0);
+  std::string content = rand.RandomString(32768);
+  Write(fname, content);
+
+  FileOptions opts;
+  std::unique_ptr<RandomAccessFileReader> r;
+  Read(fname, opts, &r);
+
+  size_t alignment = r->file()->GetRequiredBufferAlignment();
+  size_t n = alignment / 2;
+
+  int read_async_called = 0;
+  SyncPoint::GetInstance()->SetCallBack(
+      "FilePrefetchBuffer::ReadAsync",
+      [&](void* /*arg*/) { read_async_called++; });
+  SyncPoint::GetInstance()->EnableProcessing();
+
+  // Without readahead enabled, there will be no alignment and offset of buffer
+  // will be n.
+  {
+    FilePrefetchBuffer fpb(
+        /*readahead_size=*/8192, /*max_readahead_size=*/16384, /*enable=*/true,
+        /*track_min_offset=*/false, /*implicit_auto_readahead=*/true,
+        /*num_file_reads=*/0, /*num_file_reads_for_auto_readahead=*/2,
+        /*upper_bound_offset=*/0, fs());
+
+    Slice result;
+    // Simulate a seek of half of alignment bytes at offset n. Due to the
+    // readahead settings, it won't prefetch extra or do any alignment and
+    // offset of buffer will be n.
+    Status s = fpb.PrefetchAsync(IOOptions(), r.get(), n, n, &result);
+
+    // Platforms that don't have IO uring may not support async IO.
+    if (s.IsNotSupported()) {
+      return;
+    }
+
+    ASSERT_TRUE(s.IsTryAgain());
+
+    IOOptions io_opts;
+    io_opts.rate_limiter_priority = Env::IOPriority::IO_LOW;
+    ASSERT_TRUE(fpb.TryReadFromCacheAsync(io_opts, r.get(), n, n, &result, &s));
+
+    if (read_async_called) {
+      ASSERT_EQ(fpb.GetPrefetchOffset(), n);
+    }
+  }
+
+  // With readahead enabled, it will do the alignment and prefetch and offset of
+  // buffer will be 0.
+  {
+    read_async_called = false;
+    FilePrefetchBuffer fpb(
+        /*readahead_size=*/16384, /*max_readahead_size=*/16384, /*enable=*/true,
+        /*track_min_offset=*/false, /*implicit_auto_readahead=*/false,
+        /*num_file_reads=*/0, /*num_file_reads_for_auto_readahead=*/2,
+        /*upper_bound_offset=*/0, fs());
+
+    Slice result;
+    // Simulate a seek of half of alignment bytes at offset n.
+    Status s = fpb.PrefetchAsync(IOOptions(), r.get(), n, n, &result);
+
+    // Platforms that don't have IO uring may not support async IO.
+    if (s.IsNotSupported()) {
+      return;
+    }
+
+    ASSERT_TRUE(s.IsTryAgain());
+
+    IOOptions io_opts;
+    io_opts.rate_limiter_priority = Env::IOPriority::IO_LOW;
+    ASSERT_TRUE(fpb.TryReadFromCacheAsync(io_opts, r.get(), n, n, &result, &s));
+
+    if (read_async_called) {
+      ASSERT_EQ(fpb.GetPrefetchOffset(), 0);
+    }
+  }
+}
+
 TEST_F(FilePrefetchBufferTest, NoSyncWithAsyncIO) {
   std::string fname = "seek-with-block-cache-hit";
   Random rand(0);

include/rocksdb/listener.h

@@ -161,6 +161,7 @@ enum class CompactionReason : int {
   kNumOfReasons,
 };
 
+// When adding flush reason, make sure to also update `GetFlushReasonString()`.
 enum class FlushReason : int {
   kOthers = 0x00,
   kGetLiveFiles = 0x01,
@@ -178,6 +179,8 @@ enum class FlushReason : int {
   // will not be called to avoid many small immutable memtables.
   kErrorRecoveryRetryFlush = 0xc,
   kWalFull = 0xd,
+  // SwitchMemtable will not be called for this flush reason.
+  kCatchUpAfterErrorRecovery = 0xe,
 };
 
 // TODO: In the future, BackgroundErrorReason will only be used to indicate

include/rocksdb/options.h

@@ -955,10 +955,10 @@ struct DBOptions {
   // running RocksDB on spinning disks, you should set this to at least 2MB.
   // That way RocksDB's compaction is doing sequential instead of random reads.
   //
-  // Default: 0
+  // Default: 2MB
   //
   // Dynamically changeable through SetDBOptions() API.
-  size_t compaction_readahead_size = 0;
+  size_t compaction_readahead_size = 2 * 1024 * 1024;
 
   // This is a maximum buffer size that is used by WinMmapReadableFile in
   // unbuffered disk I/O mode. We need to maintain an aligned buffer for

include/rocksdb/version.h

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

table/block_based/block_based_table_iterator.cc

@@ -79,13 +79,20 @@ void BlockBasedTableIterator::SeekImpl(const Slice* target,
     }
   }
 
-  if (read_options_.auto_readahead_size && read_options_.iterate_upper_bound) {
+  if (read_options_.auto_readahead_size && read_options_.iterate_upper_bound &&
+      is_first_pass) {
     FindReadAheadSizeUpperBound();
     if (target) {
       index_iter_->Seek(*target);
     } else {
       index_iter_->SeekToFirst();
     }
+
+    // Check for IO error.
+    if (!index_iter_->Valid()) {
+      ResetDataIter();
+      return;
+    }
   }
 
   IndexValue v = index_iter_->value();

table/block_based/block_prefetcher.cc

@@ -22,7 +22,7 @@ void BlockPrefetcher::PrefetchIfNeeded(const BlockBasedTable::Rep* rep,
   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_) {
@@ -37,11 +37,12 @@ void BlockPrefetcher::PrefetchIfNeeded(const BlockBasedTable::Rep* rep,
       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.
@@ -123,8 +124,6 @@ void BlockPrefetcher::PrefetchIfNeeded(const BlockBasedTable::Rep* rep,
   }
 
   // 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()) {

table/compaction_merging_iterator.cc

@@ -329,6 +329,7 @@ void CompactionMergingIterator::FindNextVisibleKey() {
       assert(current->iter.status().ok());
       minHeap_.replace_top(current);
     } else {
+      considerStatus(current->iter.status());
       minHeap_.pop();
     }
     if (range_tombstone_iters_[current->level]) {

table/merging_iterator.cc

@@ -308,6 +308,7 @@ class MergingIterator : public InternalIterator {
     // holds after this call, and minHeap_.top().iter points to the
     // first key >= target among children_ that is not covered by any range
     // tombstone.
+    status_ = Status::OK();
     SeekImpl(target);
     FindNextVisibleKey();
 
@@ -321,6 +322,7 @@ class MergingIterator : public InternalIterator {
   void SeekForPrev(const Slice& target) override {
     assert(range_tombstone_iters_.empty() ||
            range_tombstone_iters_.size() == children_.size());
+    status_ = Status::OK();
     SeekForPrevImpl(target);
     FindPrevVisibleKey();
 
@@ -798,7 +800,6 @@ void MergingIterator::SeekImpl(const Slice& target, size_t starting_level,
     active_.erase(active_.lower_bound(starting_level), active_.end());
   }
 
-  status_ = Status::OK();
   IterKey current_search_key;
   current_search_key.SetInternalKey(target, false /* copy */);
   // Seek target might change to some range tombstone end key, so
@@ -931,6 +932,7 @@ bool MergingIterator::SkipNextDeleted() {
       InsertRangeTombstoneToMinHeap(current->level, true /* start_key */,
                                     true /* replace_top */);
     } else {
+      // TruncatedRangeDelIterator does not have status
       minHeap_.pop();
     }
     return true /* current key deleted */;
@@ -988,6 +990,9 @@ bool MergingIterator::SkipNextDeleted() {
     if (current->iter.Valid()) {
       assert(current->iter.status().ok());
       minHeap_.push(current);
+    } else {
+      // TODO(cbi): check status and early return if non-ok.
+      considerStatus(current->iter.status());
     }
     // Invariants (rti) and (phi)
     if (range_tombstone_iters_[current->level] &&
@@ -1027,6 +1032,7 @@ bool MergingIterator::SkipNextDeleted() {
         if (current->iter.Valid()) {
           minHeap_.replace_top(current);
         } else {
+          considerStatus(current->iter.status());
           minHeap_.pop();
         }
         return true /* current key deleted */;
@@ -1078,7 +1084,6 @@ void MergingIterator::SeekForPrevImpl(const Slice& target,
     active_.erase(active_.lower_bound(starting_level), active_.end());
   }
 
-  status_ = Status::OK();
   IterKey current_search_key;
   current_search_key.SetInternalKey(target, false /* copy */);
   // Seek target might change to some range tombstone end key, so
@@ -1199,6 +1204,8 @@ bool MergingIterator::SkipPrevDeleted() {
     if (current->iter.Valid()) {
       assert(current->iter.status().ok());
       maxHeap_->push(current);
+    } else {
+      considerStatus(current->iter.status());
     }
 
     if (range_tombstone_iters_[current->level] &&
@@ -1241,6 +1248,7 @@ bool MergingIterator::SkipPrevDeleted() {
         if (current->iter.Valid()) {
           maxHeap_->replace_top(current);
         } else {
+          considerStatus(current->iter.status());
           maxHeap_->pop();
         }
         return true /* current key deleted */;

unreleased_history/behavior_changes/more_sst_read_micros.md

@@ -1 +0,0 @@
-Statistics `rocksdb.sst.read.micros` now includes time spent on multi read and async read into the file

unreleased_history/behavior_changes/universal_ttl_periodic_compaction.md

@@ -1 +0,0 @@
-For Universal Compaction users, periodic compaction (option `periodic_compaction_seconds`) will be set to 30 days by default if block based table is used.

unreleased_history/bug_fixes/shifting_componeng_too_large_file_ttl_booster.md

@@ -1 +0,0 @@
-Fix a bug in FileTTLBooster that can cause users with a large number of levels (more than 65) to see errors like "runtime error: shift exponent .. is too large.." (#11673).

unreleased_history/new_features/1_context_checksum.md

@@ -1 +0,0 @@
-* 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.

unreleased_history/new_features/auto_readahead.md

@@ -1 +0,0 @@
-Add a new feature to trim readahead_size during scans upto upper_bound when iterate_upper_bound is specified. It's enabled through ReadOptions.auto_readahead_size. Users must also specify ReadOptions.iterate_upper_bound.

unreleased_history/new_features/compaction_verify_input_count.md

@@ -1 +0,0 @@
-* RocksDB will compare the number of input keys to the number of keys processed after each compaction. Compaction will fail and report Corruption status if the verification fails. Option `compaction_verify_record_count` is introduced for this purpose and is enabled by default.

unreleased_history/new_features/delay_bottommost_single_file_compaction.md

@@ -1 +0,0 @@
-Add a CF option `bottommost_file_compaction_delay` to allow specifying the delay of bottommost level single-file compactions.

unreleased_history/new_features/enable_disable_udt.md

@@ -1 +0,0 @@
-Add support to allow enabling / disabling user-defined timestamps feature for an existing column family in combination with the in-Memtable only feature.

unreleased_history/new_features/implement_allow_cache_hit_admission_policy.md

@@ -1 +0,0 @@
-Implement a new admission policy for the compressed secondary cache that admits blocks evicted from the primary cache with the hit bit set. This policy can be specified in TieredVolatileCacheOptions by setting the newly added adm_policy option.

unreleased_history/new_features/memetable_range_del_limit.md

@@ -1 +0,0 @@
-Add a column family option `memtable_max_range_deletions` that limits the number of range deletions in a memtable. RocksDB will try to do an automatic flush after the limit is reached. (#11358)

unreleased_history/new_features/put_entity_support_in_sst_file_writer.md

@@ -1 +0,0 @@
-Add PutEntity API in sst_file_writer

unreleased_history/new_features/timeout_for_wait_for_compact_api.md

@@ -1 +0,0 @@
-Add `timeout` in microsecond option to `WaitForCompactOptions` to allow timely termination of prolonged waiting in scenarios like recurring recoverable errors, such as out-of-space situations and continuous write streams that sustain ongoing flush and compactions

unreleased_history/new_features/user_read_io_activity_stats.md

@@ -1 +0,0 @@
-New statistics `rocksdb.file.read.{db.open|get|multiget|db.iterator|verify.checksum|verify.file.checksums}.micros` measure read time of block-based SST tables or blob files during db open, `Get()`, `MultiGet()`, using db iterator, `VerifyFileChecksums()` and `VerifyChecksum()`. They require stats level greater than `StatsLevel::kExceptDetailedTimers`.

unreleased_history/new_features/wait_for_compact_close_db_option.md

@@ -1 +0,0 @@
-Add close_db option to `WaitForCompactOptions` to call Close() after waiting is done.

unreleased_history/new_features/zstd-checksum.md

@@ -1 +0,0 @@
-* Add a new compression option `CompressionOptions::checksum` for enabling ZSTD's checksum feature to detect corruption during decompression.

unreleased_history/public_api_changes/mark_dep_access_hint_on_compaction_start.md

@@ -1 +0,0 @@
-Mark `Options::access_hint_on_compaction_start` related APIs as deprecated. See #11631 for alternative behavior.