Update HISTORY.md and version.h to 7.2.2

Summary:
In FilePrefetchBuffer, in case data is overlapping between two
buffers and more data is required to read and copy that to third buffer,
incorrect length was updated resulting in
```
Iterator diverged from control iterator which has value 00000000000310C3000000000000012B0000000000000274 total_order_seek: 1 auto_prefix_mode: 0 S 000000000002C37F000000000000012B000000000000001C NNNPPPPPNN; total_order_seek: 1 auto_prefix_mode: 0 S 000000000002F10B00000000000000BF78787878787878 NNNPNNNNPN; total_order_seek: 1 auto_prefix_mode: 0 S 00000000000310C3000000000000012B000000000000026B
iterator is not valid
Control CF default
db_stress: db_stress_tool/db_stress_test_base.cc:1388: void rocksdb::StressTest::VerifyIterator(rocksdb::ThreadState*, rocksdb::ColumnFamilyHandle*, const rocksdb::ReadOptions&, rocksdb::Iterator*, rocksdb::Iterator*, rocksdb::StressTest::LastIterateOp, const rocksdb::Slice&, const string&, bool*): Assertion `false' failed.
Aborted (core dumped)
```

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

Test Plan:
```
- CircleCI jobs
- Ran db_stress with OPTIONS file which caught the bug
 ./db_stress --acquire_snapshot_one_in=10000 --adaptive_readahead=0 --async_io=1 --avoid_flush_during_recovery=0 --avoid_unnecessary_blocking_io=1 --backup_max_size=104857600 --backup_one_in=0 --batch_protection_bytes_per_key=0 --block_size=16384 --bloom_bits=42.26248932628998 --bottommost_compression_type=disable --cache_index_and_filter_blocks=0 --cache_size=8388608 --checkpoint_one_in=0 --checksum_type=kxxHash --clear_column_family_one_in=0 --compact_files_one_in=1000000 --compact_range_one_in=1000000 --compaction_ttl=0 --compression_max_dict_buffer_bytes=1073741823 --compression_max_dict_bytes=16384 --compression_parallel_threads=1 --compression_type=zstd --compression_zstd_max_train_bytes=65536 --continuous_verification_interval=0 --db=/dev/shm/rocksdb/ --db_write_buffer_size=134217728 --delpercent=5 --delrangepercent=0 --destroy_db_initially=0 --detect_filter_construct_corruption=0 --disable_wal=0 --enable_blob_files=0 --enable_compaction_filter=0 --enable_pipelined_write=0 --fail_if_options_file_error=0 --file_checksum_impl=none --flush_one_in=1000000 --format_version=4 --get_current_wal_file_one_in=0 --get_live_files_one_in=1000000 --get_property_one_in=1000000 --get_sorted_wal_files_one_in=0 --index_block_restart_interval=12 --index_type=2 --ingest_external_file_one_in=0 --iterpercent=10 --key_len_percent_dist=1,30,69 --level_compaction_dynamic_level_bytes=True --long_running_snapshots=0 --mark_for_compaction_one_file_in=0 --max_background_compactions=20 --max_bytes_for_level_base=10485760 --max_key=25000000 --max_key_len=3 --max_manifest_file_size=1073741824 --max_write_batch_group_size_bytes=1048576 --max_write_buffer_number=3 --max_write_buffer_size_to_maintain=8388608 --memtable_prefix_bloom_size_ratio=0.001 --memtable_whole_key_filtering=1 --memtablerep=skip_list --mmap_read=0 --mock_direct_io=False --nooverwritepercent=1 --open_files=100 --open_metadata_write_fault_one_in=0 --open_read_fault_one_in=0 --open_write_fault_one_in=16 --ops_per_thread=100000000 --optimize_filters_for_memory=0 --paranoid_file_checks=1 --partition_filters=0 --partition_pinning=1 --pause_background_one_in=1000000 --periodic_compaction_seconds=0 --prefix_size=-1 --prefixpercent=0 --prepopulate_block_cache=0 --progress_reports=0 --read_fault_one_in=0 --read_only=0 --readpercent=50 --recycle_log_file_num=1 --reopen=0 --reserve_table_reader_memory=0 --ribbon_starting_level=999 --secondary_cache_fault_one_in=0 --secondary_catch_up_one_in=0 --set_options_one_in=10000 --snapshot_hold_ops=100000 --sst_file_manager_bytes_per_sec=104857600 --sst_file_manager_bytes_per_truncate=0 --subcompactions=2 --sync=0 --sync_fault_injection=False --target_file_size_base=2097152 --target_file_size_multiplier=2 --test_batches_snapshots=0 --test_cf_consistency=0 --top_level_index_pinning=3 --unpartitioned_pinning=3 --use_blob_db=0 --use_block_based_filter=0 --use_clock_cache=0 --use_direct_io_for_flush_and_compaction=1 --use_direct_reads=0 --use_full_merge_v1=0 --use_merge=0 --use_multiget=0 --use_txn=0 --user_timestamp_size=0 --value_size_mult=32 --verify_checksum=1 --verify_checksum_one_in=1000000 --verify_db_one_in=100000 --wal_compression=zstd --write_buffer_size=4194304 --write_dbid_to_manifest=0 --writepercent=35 --options_file=/home/akankshamahajan/OPTIONS.orig -column_families=1

db_bench with async_io enabled to make sure db_bench completes successfully without any failure.
- ./db_bench -use_existing_db=true -db=/tmp/prefix_scan_prefetch_main -benchmarks="seekrandom" -key_size=32 -value_size=512 -num=5000000 -use_direct_reads=true -seek_nexts=327680 -duration=120 -ops_between_duration_checks=1 -async_io=1
```

crash_test in progress

Reviewed By: anand1976

Differential Revision: D35985789

Pulled By: akankshamahajan15

fbshipit-source-id: 5abe185f34caa99ca587d4bdc8954bd0802b1bf9

HISTORY.md

@@ -1,4 +1,13 @@
 # Rocksdb Change Log
+## 7.2.2 (04/28/2022)
+### Bug Fixes
+* Fixed a bug in async_io path where incorrect length of data is read by FilePrefetchBuffer if data is consumed from two populated buffers and request for more data is sent.
+
+## 7.2.1 (04/26/2022)
+### Bug Fixes
+* Fixed a bug where RocksDB could corrupt DBs with `avoid_flush_during_recovery == true` by removing valid WALs, leading to `Status::Corruption` with message like "SST file is ahead of WALs" when attempting to reopen.
+* RocksDB calls FileSystem::Poll API during FilePrefetchBuffer destruction which impacts performance as it waits for read requets completion which is not needed anymore. Calling FileSystem::AbortIO to abort those requests instead fixes that performance issue.
+
 ## 7.2.0 (04/15/2022)
 ### Bug Fixes
 * Fixed bug which caused rocksdb failure in the situation when rocksdb was accessible using UNC path
@@ -22,6 +31,8 @@
 * Added a dedicated integer DB property `rocksdb.live-blob-file-garbage-size` that exposes the total amount of garbage in the blob files in the current version.
 * RocksDB does internal auto prefetching if it notices sequential reads. It starts with readahead size `initial_auto_readahead_size` which now can be configured through BlockBasedTableOptions.
 * Add a merge operator that allows users to register specific aggregation function so that they can does aggregation using different aggregation types for different keys. See comments in include/rocksdb/utilities/agg_merge.h for actual usage. The feature is experimental and the format is subject to change and we won't provide a migration tool.
+* Meta-internal / Experimental: Improve CPU performance by replacing many uses of std::unordered_map with folly::F14FastMap when RocksDB is compiled together with Folly.
+* Experimental: Add CompressedSecondaryCache, a concrete implementation of rocksdb::SecondaryCache, that integrates with compression libraries (e.g. LZ4) to hold compressed blocks.
 
 ### Behavior changes
 * Disallow usage of commit-time-write-batch for write-prepared/write-unprepared transactions if TransactionOptions::use_only_the_last_commit_time_batch_for_recovery is false to prevent two (or more) uncommitted versions of the same key in the database. Otherwise, bottommost compaction may violate the internal key uniqueness invariant of SSTs if the sequence numbers of both internal keys are zeroed out (#9794).
@@ -30,6 +41,7 @@
 ### Public API changes
 * Exposed APIs to examine results of block cache stats collections in a structured way. In particular, users of `GetMapProperty()` with property `kBlockCacheEntryStats` can now use the functions in `BlockCacheEntryStatsMapKeys` to find stats in the map.
 * Add `fail_if_not_bottommost_level` to IngestExternalFileOptions so that ingestion will fail if the file(s) cannot be ingested to the bottommost level.
+* Add output parameter `is_in_sec_cache` to `SecondaryCache::Lookup()`. It is to indicate whether the handle is possibly erased from the secondary cache after the Lookup.
 
 ## 7.1.0 (03/23/2022)
 ### New Features

db/corruption_test.cc

@@ -337,6 +337,68 @@ TEST_F(CorruptionTest, Recovery) {
   Check(36, 36);
 }
 
+TEST_F(CorruptionTest, PostPITRCorruptionWALsRetained) {
+  // Repro for bug where WALs following the point-in-time recovery were not
+  // retained leading to the next recovery failing.
+  CloseDb();
+
+  options_.wal_recovery_mode = WALRecoveryMode::kPointInTimeRecovery;
+
+  const std::string test_cf_name = "test_cf";
+  std::vector<ColumnFamilyDescriptor> cf_descs;
+  cf_descs.emplace_back(kDefaultColumnFamilyName, ColumnFamilyOptions());
+  cf_descs.emplace_back(test_cf_name, ColumnFamilyOptions());
+
+  uint64_t log_num;
+  {
+    options_.create_missing_column_families = true;
+    std::vector<ColumnFamilyHandle*> cfhs;
+    ASSERT_OK(DB::Open(options_, dbname_, cf_descs, &cfhs, &db_));
+
+    ASSERT_OK(db_->Put(WriteOptions(), cfhs[0], "k", "v"));
+    ASSERT_OK(db_->Put(WriteOptions(), cfhs[1], "k", "v"));
+    ASSERT_OK(db_->Put(WriteOptions(), cfhs[0], "k2", "v2"));
+    std::vector<uint64_t> file_nums;
+    GetSortedWalFiles(file_nums);
+    log_num = file_nums.back();
+    for (auto* cfh : cfhs) {
+      delete cfh;
+    }
+    CloseDb();
+  }
+
+  CorruptFileWithTruncation(FileType::kWalFile, log_num,
+                            /*bytes_to_truncate=*/1);
+
+  {
+    // Recover "k" -> "v" for both CFs. "k2" -> "v2" is lost due to truncation.
+    options_.avoid_flush_during_recovery = true;
+    std::vector<ColumnFamilyHandle*> cfhs;
+    ASSERT_OK(DB::Open(options_, dbname_, cf_descs, &cfhs, &db_));
+    // Flush one but not both CFs and write some data so there's a seqno gap
+    // between the PITR corruption and the next DB session's first WAL.
+    ASSERT_OK(db_->Put(WriteOptions(), cfhs[1], "k2", "v2"));
+    ASSERT_OK(db_->Flush(FlushOptions(), cfhs[1]));
+
+    for (auto* cfh : cfhs) {
+      delete cfh;
+    }
+    CloseDb();
+  }
+
+  // With the bug, this DB open would remove the WALs following the PITR
+  // corruption. Then, the next recovery would fail.
+  for (int i = 0; i < 2; ++i) {
+    std::vector<ColumnFamilyHandle*> cfhs;
+    ASSERT_OK(DB::Open(options_, dbname_, cf_descs, &cfhs, &db_));
+
+    for (auto* cfh : cfhs) {
+      delete cfh;
+    }
+    CloseDb();
+  }
+}
+
 TEST_F(CorruptionTest, RecoverWriteError) {
   env_->writable_file_error_ = true;
   Status s = TryReopen();

db/db_impl/db_impl.h

@@ -1240,43 +1240,6 @@ class DBImpl : public DB {
 
   std::atomic<bool> shutting_down_;
 
-  // RecoveryContext struct stores the context about version edits along
-  // with corresponding column_family_data and column_family_options.
-  class RecoveryContext {
-   public:
-    ~RecoveryContext() {
-      for (auto& edit_list : edit_lists_) {
-        for (auto* edit : edit_list) {
-          delete edit;
-        }
-        edit_list.clear();
-      }
-      cfds_.clear();
-      mutable_cf_opts_.clear();
-      edit_lists_.clear();
-      files_to_delete_.clear();
-    }
-
-    void UpdateVersionEdits(ColumnFamilyData* cfd, const VersionEdit& edit) {
-      if (map_.find(cfd->GetID()) == map_.end()) {
-        uint32_t size = static_cast<uint32_t>(map_.size());
-        map_.emplace(cfd->GetID(), size);
-        cfds_.emplace_back(cfd);
-        mutable_cf_opts_.emplace_back(cfd->GetLatestMutableCFOptions());
-        edit_lists_.emplace_back(autovector<VersionEdit*>());
-      }
-      uint32_t i = map_[cfd->GetID()];
-      edit_lists_[i].emplace_back(new VersionEdit(edit));
-    }
-
-    std::unordered_map<uint32_t, uint32_t> map_;  // cf_id to index;
-    autovector<ColumnFamilyData*> cfds_;
-    autovector<const MutableCFOptions*> mutable_cf_opts_;
-    autovector<autovector<VersionEdit*>> edit_lists_;
-    // files_to_delete_ contains sst files
-    std::set<std::string> files_to_delete_;
-  };
-
   // Except in DB::Open(), WriteOptionsFile can only be called when:
   // Persist options to options file.
   // If need_mutex_lock = false, the method will lock DB mutex.
@@ -1393,19 +1356,16 @@ class DBImpl : public DB {
   // be made to the descriptor are added to *edit.
   // recovered_seq is set to less than kMaxSequenceNumber if the log's tail is
   // skipped.
-  // recovery_ctx stores the context about version edits and all those
-  // edits are persisted to new Manifest after successfully syncing the new WAL.
   virtual Status Recover(
       const std::vector<ColumnFamilyDescriptor>& column_families,
       bool read_only = false, bool error_if_wal_file_exists = false,
       bool error_if_data_exists_in_wals = false,
-      uint64_t* recovered_seq = nullptr,
-      RecoveryContext* recovery_ctx = nullptr);
+      uint64_t* recovered_seq = nullptr);
 
   virtual bool OwnTablesAndLogs() const { return true; }
 
   // Set DB identity file, and write DB ID to manifest if necessary.
-  Status SetDBId(bool read_only, RecoveryContext* recovery_ctx);
+  Status SetDBId(bool read_only);
 
   // REQUIRES: db mutex held when calling this function, but the db mutex can
   // be released and re-acquired. Db mutex will be held when the function
@@ -1414,15 +1374,12 @@ class DBImpl : public DB {
   // not referenced in the MANIFEST (e.g.
   // 1. It's best effort recovery;
   // 2. The VersionEdits referencing the SST files are appended to
-  // RecoveryContext, DB crashes when syncing the MANIFEST, the VersionEdits are
+  // MANIFEST, DB crashes when syncing the MANIFEST, the VersionEdits are
   // still not synced to MANIFEST during recovery.)
-  // It stores the SST files to be deleted in RecoveryContext. In the
+  // We delete these SST files. In the
   // meantime, we find out the largest file number present in the paths, and
   // bump up the version set's next_file_number_ to be 1 + largest_file_number.
-  // recovery_ctx stores the context about version edits and files to be
-  // deleted. All those edits are persisted to new Manifest after successfully
-  // syncing the new WAL.
-  Status DeleteUnreferencedSstFiles(RecoveryContext* recovery_ctx);
+  Status DeleteUnreferencedSstFiles();
 
   // SetDbSessionId() should be called in the constuctor DBImpl()
   // to ensure that db_session_id_ gets updated every time the DB is opened
@@ -1432,11 +1389,6 @@ class DBImpl : public DB {
   Status FailIfTsSizesMismatch(const ColumnFamilyHandle* column_family,
                                const Slice& ts) const;
 
-  // recovery_ctx stores the context about version edits and
-  // LogAndApplyForRecovery persist all those edits to new Manifest after
-  // successfully syncing new WAL.
-  Status LogAndApplyForRecovery(const RecoveryContext& recovery_ctx);
-
  private:
   friend class DB;
   friend class ErrorHandler;
@@ -1691,10 +1643,9 @@ class DBImpl : public DB {
 
   // REQUIRES: log_numbers are sorted in ascending order
   // corrupted_log_found is set to true if we recover from a corrupted log file.
-  Status RecoverLogFiles(std::vector<uint64_t>& log_numbers,
+  Status RecoverLogFiles(const std::vector<uint64_t>& log_numbers,
                          SequenceNumber* next_sequence, bool read_only,
-                         bool* corrupted_log_found,
-                         RecoveryContext* recovery_ctx);
+                         bool* corrupted_log_found);
 
   // The following two methods are used to flush a memtable to
   // storage. The first one is used at database RecoveryTime (when the
@@ -1704,12 +1655,6 @@ class DBImpl : public DB {
   Status WriteLevel0TableForRecovery(int job_id, ColumnFamilyData* cfd,
                                      MemTable* mem, VersionEdit* edit);
 
-  // Move all the WAL files starting from corrupted WAL found to
-  // max_wal_number to avoid column family inconsistency error on recovery. It
-  // also removes the deleted file from the vector wal_numbers.
-  void MoveCorruptedWalFiles(std::vector<uint64_t>& wal_numbers,
-                             uint64_t corrupted_wal_number);
-
   // Get the size of a log file and, if truncate is true, truncate the
   // log file to its actual size, thereby freeing preallocated space.
   // Return success even if truncate fails

db/db_impl/db_impl_files.cc

@@ -863,7 +863,7 @@ uint64_t PrecomputeMinLogNumberToKeep2PC(
   return min_log_number_to_keep;
 }
 
-Status DBImpl::SetDBId(bool read_only, RecoveryContext* recovery_ctx) {
+Status DBImpl::SetDBId(bool read_only) {
   Status s;
   // Happens when immutable_db_options_.write_dbid_to_manifest is set to true
   // the very first time.
@@ -890,14 +890,14 @@ Status DBImpl::SetDBId(bool read_only, RecoveryContext* recovery_ctx) {
     }
     s = GetDbIdentityFromIdentityFile(&db_id_);
     if (immutable_db_options_.write_dbid_to_manifest && s.ok()) {
-      assert(!read_only);
-      assert(recovery_ctx != nullptr);
-      assert(versions_->GetColumnFamilySet() != nullptr);
       VersionEdit edit;
       edit.SetDBId(db_id_);
+      Options options;
+      MutableCFOptions mutable_cf_options(options);
       versions_->db_id_ = db_id_;
-      recovery_ctx->UpdateVersionEdits(
-          versions_->GetColumnFamilySet()->GetDefault(), edit);
+      s = versions_->LogAndApply(versions_->GetColumnFamilySet()->GetDefault(),
+                                 mutable_cf_options, &edit, &mutex_, nullptr,
+                                 /* new_descriptor_log */ false);
     }
   } else if (!read_only) {
     s = SetIdentityFile(env_, dbname_, db_id_);
@@ -905,7 +905,7 @@ Status DBImpl::SetDBId(bool read_only, RecoveryContext* recovery_ctx) {
   return s;
 }
 
-Status DBImpl::DeleteUnreferencedSstFiles(RecoveryContext* recovery_ctx) {
+Status DBImpl::DeleteUnreferencedSstFiles() {
   mutex_.AssertHeld();
   std::vector<std::string> paths;
   paths.push_back(NormalizePath(dbname_ + std::string(1, kFilePathSeparator)));
@@ -925,6 +925,7 @@ Status DBImpl::DeleteUnreferencedSstFiles(RecoveryContext* recovery_ctx) {
 
   uint64_t next_file_number = versions_->current_next_file_number();
   uint64_t largest_file_number = next_file_number;
+  std::set<std::string> files_to_delete;
   Status s;
   for (const auto& path : paths) {
     std::vector<std::string> files;
@@ -942,9 +943,8 @@ Status DBImpl::DeleteUnreferencedSstFiles(RecoveryContext* recovery_ctx) {
       const std::string normalized_fpath = path + fname;
       largest_file_number = std::max(largest_file_number, number);
       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_.insert(normalized_fpath);
+          files_to_delete.find(normalized_fpath) == files_to_delete.end()) {
+        files_to_delete.insert(normalized_fpath);
       }
     }
   }
@@ -961,7 +961,21 @@ Status DBImpl::DeleteUnreferencedSstFiles(RecoveryContext* recovery_ctx) {
   assert(versions_->GetColumnFamilySet());
   ColumnFamilyData* default_cfd = versions_->GetColumnFamilySet()->GetDefault();
   assert(default_cfd);
-  recovery_ctx->UpdateVersionEdits(default_cfd, edit);
+  s = versions_->LogAndApply(
+      default_cfd, *default_cfd->GetLatestMutableCFOptions(), &edit, &mutex_,
+      directories_.GetDbDir(), /*new_descriptor_log*/ false);
+  if (!s.ok()) {
+    return s;
+  }
+
+  mutex_.Unlock();
+  for (const auto& fname : files_to_delete) {
+    s = env_->DeleteFile(fname);
+    if (!s.ok()) {
+      break;
+    }
+  }
+  mutex_.Lock();
   return s;
 }
 

db/db_impl/db_impl_open.cc

@@ -399,7 +399,7 @@ IOStatus Directories::SetDirectories(FileSystem* fs, const std::string& dbname,
 Status DBImpl::Recover(
     const std::vector<ColumnFamilyDescriptor>& column_families, bool read_only,
     bool error_if_wal_file_exists, bool error_if_data_exists_in_wals,
-    uint64_t* recovered_seq, RecoveryContext* recovery_ctx) {
+    uint64_t* recovered_seq) {
   mutex_.AssertHeld();
 
   bool is_new_db = false;
@@ -518,10 +518,9 @@ Status DBImpl::Recover(
   if (!s.ok()) {
     return s;
   }
-
-  s = SetDBId(read_only, recovery_ctx);
+  s = SetDBId(read_only);
   if (s.ok() && !read_only) {
-    s = DeleteUnreferencedSstFiles(recovery_ctx);
+    s = DeleteUnreferencedSstFiles();
   }
 
   if (immutable_db_options_.paranoid_checks && s.ok()) {
@@ -536,6 +535,10 @@ Status DBImpl::Recover(
       }
     }
   }
+  // DB mutex is already held
+  if (s.ok() && immutable_db_options_.persist_stats_to_disk) {
+    s = InitPersistStatsColumnFamily();
+  }
 
   std::vector<std::string> files_in_wal_dir;
   if (s.ok()) {
@@ -605,10 +608,7 @@ Status DBImpl::Recover(
       WalNumber max_wal_number =
           versions_->GetWalSet().GetWals().rbegin()->first;
       edit.DeleteWalsBefore(max_wal_number + 1);
-      assert(recovery_ctx != nullptr);
-      assert(versions_->GetColumnFamilySet() != nullptr);
-      recovery_ctx->UpdateVersionEdits(
-          versions_->GetColumnFamilySet()->GetDefault(), edit);
+      s = versions_->LogAndApplyToDefaultColumnFamily(&edit, &mutex_);
     }
     if (!s.ok()) {
       return s;
@@ -644,8 +644,8 @@ Status DBImpl::Recover(
       std::sort(wals.begin(), wals.end());
 
       bool corrupted_wal_found = false;
-      s = RecoverLogFiles(wals, &next_sequence, read_only, &corrupted_wal_found,
-                          recovery_ctx);
+      s = RecoverLogFiles(wals, &next_sequence, read_only,
+                          &corrupted_wal_found);
       if (corrupted_wal_found && recovered_seq != nullptr) {
         *recovered_seq = next_sequence;
       }
@@ -805,30 +805,10 @@ Status DBImpl::InitPersistStatsColumnFamily() {
   return s;
 }
 
-Status DBImpl::LogAndApplyForRecovery(const RecoveryContext& recovery_ctx) {
-  mutex_.AssertHeld();
-  assert(versions_->descriptor_log_ == nullptr);
-  Status s = versions_->LogAndApply(
-      recovery_ctx.cfds_, recovery_ctx.mutable_cf_opts_,
-      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);
-      if (!s.ok()) {
-        break;
-      }
-    }
-    mutex_.Lock();
-  }
-  return s;
-}
-
 // REQUIRES: wal_numbers are sorted in ascending order
-Status DBImpl::RecoverLogFiles(std::vector<uint64_t>& wal_numbers,
+Status DBImpl::RecoverLogFiles(const std::vector<uint64_t>& wal_numbers,
                                SequenceNumber* next_sequence, bool read_only,
-                               bool* corrupted_wal_found,
-                               RecoveryContext* recovery_ctx) {
+                               bool* corrupted_wal_found) {
   struct LogReporter : public log::Reader::Reporter {
     Env* env;
     Logger* info_log;
@@ -853,7 +833,6 @@ Status DBImpl::RecoverLogFiles(std::vector<uint64_t>& wal_numbers,
     edit.SetColumnFamily(cfd->GetID());
     version_edits.insert({cfd->GetID(), edit});
   }
-
   int job_id = next_job_id_.fetch_add(1);
   {
     auto stream = event_logger_.Log();
@@ -1277,7 +1256,6 @@ Status DBImpl::RecoverLogFiles(std::vector<uint64_t>& wal_numbers,
         edit->SetLogNumber(max_wal_number + 1);
       }
     }
-
     if (status.ok()) {
       // we must mark the next log number as used, even though it's
       // not actually used. that is because VersionSet assumes
@@ -1285,40 +1263,42 @@ Status DBImpl::RecoverLogFiles(std::vector<uint64_t>& wal_numbers,
       // log number
       versions_->MarkFileNumberUsed(max_wal_number + 1);
 
-      if (corrupted_wal_found != nullptr && *corrupted_wal_found == true &&
-          immutable_db_options_.wal_recovery_mode ==
-              WALRecoveryMode::kPointInTimeRecovery) {
-        MoveCorruptedWalFiles(wal_numbers, corrupted_wal_number);
-      }
-
-      assert(recovery_ctx != nullptr);
+      autovector<ColumnFamilyData*> cfds;
+      autovector<const MutableCFOptions*> cf_opts;
+      autovector<autovector<VersionEdit*>> edit_lists;
       for (auto* cfd : *versions_->GetColumnFamilySet()) {
+        cfds.push_back(cfd);
+        cf_opts.push_back(cfd->GetLatestMutableCFOptions());
         auto iter = version_edits.find(cfd->GetID());
         assert(iter != version_edits.end());
-        recovery_ctx->UpdateVersionEdits(cfd, iter->second);
+        edit_lists.push_back({&iter->second});
       }
 
+      std::unique_ptr<VersionEdit> wal_deletion;
       if (flushed) {
-        VersionEdit wal_deletion;
+        wal_deletion = std::make_unique<VersionEdit>();
         if (immutable_db_options_.track_and_verify_wals_in_manifest) {
-          wal_deletion.DeleteWalsBefore(max_wal_number + 1);
+          wal_deletion->DeleteWalsBefore(max_wal_number + 1);
         }
         if (!allow_2pc()) {
           // In non-2pc mode, flushing the memtables of the column families
           // means we can advance min_log_number_to_keep.
-          wal_deletion.SetMinLogNumberToKeep(max_wal_number + 1);
+          wal_deletion->SetMinLogNumberToKeep(max_wal_number + 1);
         }
-        assert(versions_->GetColumnFamilySet() != nullptr);
-        recovery_ctx->UpdateVersionEdits(
-            versions_->GetColumnFamilySet()->GetDefault(), wal_deletion);
+        edit_lists.back().push_back(wal_deletion.get());
       }
+
+      // write MANIFEST with update
+      status = versions_->LogAndApply(cfds, cf_opts, edit_lists, &mutex_,
+                                      directories_.GetDbDir(),
+                                      /*new_descriptor_log=*/true);
     }
   }
 
   if (status.ok()) {
     if (data_seen && !flushed) {
       status = RestoreAliveLogFiles(wal_numbers);
-    } else if (!wal_numbers.empty()) {
+    } else {
       // If there's no data in the WAL, or we flushed all the data, still
       // truncate the log file. If the process goes into a crash loop before
       // the file is deleted, the preallocated space will never get freed.
@@ -1334,48 +1314,6 @@ Status DBImpl::RecoverLogFiles(std::vector<uint64_t>& wal_numbers,
   return status;
 }
 
-void DBImpl::MoveCorruptedWalFiles(std::vector<uint64_t>& wal_numbers,
-                                   uint64_t corrupted_wal_number) {
-  size_t num_wals = wal_numbers.size();
-  // Find the first corrupted wal.
-  auto iter = std::lower_bound(wal_numbers.begin(), wal_numbers.end(),
-                               corrupted_wal_number);
-  auto corrupt_start_iter = iter;
-
-  // Increment iter to move WAL files from first corrupted_wal_number + 1.
-  iter++;
-
-  std::string archival_path =
-      ArchivalDirectory(immutable_db_options_.GetWalDir());
-  Status create_status = env_->CreateDirIfMissing(archival_path);
-
-  // create_status is only checked when it needs to move the corrupted WAL files
-  // to archive folder.
-  create_status.PermitUncheckedError();
-
-  // Truncate the last WAL to reclaim the pre allocated space before
-  // moving it.
-  GetLogSizeAndMaybeTruncate(wal_numbers.back(), /*truncate=*/true, nullptr)
-      .PermitUncheckedError();
-
-  // Move all the WAL files from corrupted_wal_number + 1 to last WAL
-  // (max_wal_number) to avoid column family inconsistency error to archival
-  // directory. If its unable to create archive dir, it will delete the
-  // corrupted WAL files.
-  // We are moving all but first corrupted WAL file to a different folder.
-  while (iter != wal_numbers.end()) {
-    LogFileNumberSize log(*iter);
-    std::string fname = LogFileName(immutable_db_options_.GetWalDir(), *iter);
-#ifndef ROCKSDB_LITE
-    if (create_status.ok()) {
-      wal_manager_.ArchiveWALFile(fname, *iter);
-    }
-#endif
-    iter++;
-  }
-  wal_numbers.erase(corrupt_start_iter + 1, wal_numbers.begin() + num_wals);
-}
-
 Status DBImpl::GetLogSizeAndMaybeTruncate(uint64_t wal_number, bool truncate,
                                           LogFileNumberSize* log_ptr) {
   LogFileNumberSize log(wal_number);
@@ -1438,8 +1376,7 @@ Status DBImpl::RestoreAliveLogFiles(const std::vector<uint64_t>& wal_numbers) {
     // log has such preallocated space, so we only truncate for the last log.
     LogFileNumberSize log;
     s = GetLogSizeAndMaybeTruncate(
-        wal_number,
-        /*truncate=*/(wal_number == wal_numbers.back()), &log);
+        wal_number, /*truncate=*/(wal_number == wal_numbers.back()), &log);
     if (!s.ok()) {
       break;
     }
@@ -1800,13 +1737,9 @@ Status DBImpl::Open(const DBOptions& db_options, const std::string& dbname,
   impl->wal_in_db_path_ = impl->immutable_db_options_.IsWalDirSameAsDBPath();
 
   impl->mutex_.Lock();
-
-  RecoveryContext recovery_ctx;
-
   // Handles create_if_missing, error_if_exists
   uint64_t recovered_seq(kMaxSequenceNumber);
-  s = impl->Recover(column_families, false, false, false, &recovered_seq,
-                    &recovery_ctx);
+  s = impl->Recover(column_families, false, false, false, &recovered_seq);
   if (s.ok()) {
     uint64_t new_log_number = impl->versions_->NewFileNumber();
     log::Writer* new_log = nullptr;
@@ -1823,6 +1756,40 @@ Status DBImpl::Open(const DBOptions& db_options, const std::string& dbname,
     }
 
     if (s.ok()) {
+      // set column family handles
+      for (auto cf : column_families) {
+        auto cfd =
+            impl->versions_->GetColumnFamilySet()->GetColumnFamily(cf.name);
+        if (cfd != nullptr) {
+          handles->push_back(
+              new ColumnFamilyHandleImpl(cfd, impl, &impl->mutex_));
+          impl->NewThreadStatusCfInfo(cfd);
+        } else {
+          if (db_options.create_missing_column_families) {
+            // missing column family, create it
+            ColumnFamilyHandle* handle;
+            impl->mutex_.Unlock();
+            s = impl->CreateColumnFamily(cf.options, cf.name, &handle);
+            impl->mutex_.Lock();
+            if (s.ok()) {
+              handles->push_back(handle);
+            } else {
+              break;
+            }
+          } else {
+            s = Status::InvalidArgument("Column family not found", cf.name);
+            break;
+          }
+        }
+      }
+    }
+    if (s.ok()) {
+      SuperVersionContext sv_context(/* create_superversion */ true);
+      for (auto cfd : *impl->versions_->GetColumnFamilySet()) {
+        impl->InstallSuperVersionAndScheduleWork(
+            cfd, &sv_context, *cfd->GetLatestMutableCFOptions());
+      }
+      sv_context.Clean();
       if (impl->two_write_queues_) {
         impl->log_write_mutex_.Lock();
       }
@@ -1835,15 +1802,14 @@ Status DBImpl::Open(const DBOptions& db_options, const std::string& dbname,
     }
     if (s.ok()) {
       // In WritePrepared there could be gap in sequence numbers. This breaks
-      // the trick we use in kPointInTimeRecovery which assumes the first seq
-      // in the log right after the corrupted log is one larger than the last
-      // seq we read from the wals. To let this trick keep working, we add a
-      // dummy entry with the expected sequence to the first log right after
-      // recovery. In non-WritePrepared case also the new log after recovery
-      // could be empty, and thus missing the consecutive seq hint to
-      // distinguish middle-log corruption to
-      // corrupted-log-remained-after-recovery. This case also will be
-      // addressed by a dummy write.
+      // the trick we use in kPointInTimeRecovery which assumes the first seq in
+      // the log right after the corrupted log is one larger than the last seq
+      // we read from the wals. To let this trick keep working, we add a dummy
+      // entry with the expected sequence to the first log right after recovery.
+      // In non-WritePrepared case also the new log after recovery could be
+      // empty, and thus missing the consecutive seq hint to distinguish
+      // middle-log corruption to corrupted-log-remained-after-recovery. This
+      // case also will be addressed by a dummy write.
       if (recovered_seq != kMaxSequenceNumber) {
         WriteBatch empty_batch;
         WriteBatchInternal::SetSequence(&empty_batch, recovered_seq);
@@ -1862,52 +1828,6 @@ Status DBImpl::Open(const DBOptions& db_options, const std::string& dbname,
       }
     }
   }
-  if (s.ok()) {
-    s = impl->LogAndApplyForRecovery(recovery_ctx);
-  }
-
-  if (s.ok() && impl->immutable_db_options_.persist_stats_to_disk) {
-    impl->mutex_.AssertHeld();
-    s = impl->InitPersistStatsColumnFamily();
-  }
-
-  if (s.ok()) {
-    // set column family handles
-    for (auto cf : column_families) {
-      auto cfd =
-          impl->versions_->GetColumnFamilySet()->GetColumnFamily(cf.name);
-      if (cfd != nullptr) {
-        handles->push_back(
-            new ColumnFamilyHandleImpl(cfd, impl, &impl->mutex_));
-        impl->NewThreadStatusCfInfo(cfd);
-      } else {
-        if (db_options.create_missing_column_families) {
-          // missing column family, create it
-          ColumnFamilyHandle* handle;
-          impl->mutex_.Unlock();
-          s = impl->CreateColumnFamily(cf.options, cf.name, &handle);
-          impl->mutex_.Lock();
-          if (s.ok()) {
-            handles->push_back(handle);
-          } else {
-            break;
-          }
-        } else {
-          s = Status::InvalidArgument("Column family not found", cf.name);
-          break;
-        }
-      }
-    }
-  }
-
-  if (s.ok()) {
-    SuperVersionContext sv_context(/* create_superversion */ true);
-    for (auto cfd : *impl->versions_->GetColumnFamilySet()) {
-      impl->InstallSuperVersionAndScheduleWork(
-          cfd, &sv_context, *cfd->GetLatestMutableCFOptions());
-    }
-    sv_context.Clean();
-  }
   if (s.ok() && impl->immutable_db_options_.persist_stats_to_disk) {
     // try to read format version
     s = impl->PersistentStatsProcessFormatVersion();
@@ -1933,8 +1853,7 @@ Status DBImpl::Open(const DBOptions& db_options, const std::string& dbname,
       if (cfd->ioptions()->merge_operator != nullptr &&
           !cfd->mem()->IsMergeOperatorSupported()) {
         s = Status::InvalidArgument(
-            "The memtable of column family %s does not support merge "
-            "operator "
+            "The memtable of column family %s does not support merge operator "
             "its options.merge_operator is non-null",
             cfd->GetName().c_str());
       }

db/db_impl/db_impl_secondary.cc

@@ -33,8 +33,7 @@ DBImplSecondary::~DBImplSecondary() {}
 Status DBImplSecondary::Recover(
     const std::vector<ColumnFamilyDescriptor>& column_families,
     bool /*readonly*/, bool /*error_if_wal_file_exists*/,
-    bool /*error_if_data_exists_in_wals*/, uint64_t*,
-    RecoveryContext* /*recovery_ctx*/) {
+    bool /*error_if_data_exists_in_wals*/, uint64_t*) {
   mutex_.AssertHeld();
 
   JobContext job_context(0);

db/db_impl/db_impl_secondary.h

@@ -81,8 +81,8 @@ class DBImplSecondary : public DBImpl {
   // and log_readers_ to facilitate future operations.
   Status Recover(const std::vector<ColumnFamilyDescriptor>& column_families,
                  bool read_only, bool error_if_wal_file_exists,
-                 bool error_if_data_exists_in_wals, uint64_t* = nullptr,
-                 RecoveryContext* recovery_ctx = nullptr) override;
+                 bool error_if_data_exists_in_wals,
+                 uint64_t* = nullptr) override;
 
   // Implementations of the DB interface
   using DB::Get;

env/fs_posix.cc

@@ -1118,6 +1118,19 @@ class PosixFileSystem : public FileSystem {
 #endif
   }
 
+  // TODO akanksha: Look into flags and see how to provide support for AbortIO
+  // in posix for IOUring requests. Currently it calls Poll to wait for requests
+  // to complete the request.
+  virtual IOStatus AbortIO(std::vector<void*>& io_handles) override {
+    IOStatus s = Poll(io_handles, io_handles.size());
+    // If Poll is not supported then it didn't submit any request and it should
+    // return OK.
+    if (s.IsNotSupported()) {
+      return IOStatus::OK();
+    }
+    return s;
+  }
+
 #if defined(ROCKSDB_IOURING_PRESENT)
   // io_uring instance
   std::unique_ptr<ThreadLocalPtr> thread_local_io_urings_;

file/file_prefetch_buffer.cc

@@ -229,6 +229,7 @@ Status FilePrefetchBuffer::PrefetchAsync(const IOOptions& opts,
     // second buffer.
     std::vector<void*> handles;
     handles.emplace_back(io_handle_);
+    StopWatch sw(clock_, stats_, POLL_WAIT_MICROS);
     fs_->Poll(handles, 1).PermitUncheckedError();
   }
 
@@ -281,7 +282,7 @@ Status FilePrefetchBuffer::PrefetchAsync(const IOOptions& opts,
           bufs_[curr_].offset_ + bufs_[curr_].buffer_.CurrentSize()) {
     offset += length;
     length = 0;
-    prefetch_size -= length;
+    prefetch_size = readahead_size;
   }
   // Data is overlapping i.e. some of the data is in curr_ buffer and remaining
   // in second buffer.
@@ -310,7 +311,8 @@ Status FilePrefetchBuffer::PrefetchAsync(const IOOptions& opts,
     // sync prefetching and copy the remaining data to third buffer in the end.
     // swap the buffers.
     curr_ = curr_ ^ 1;
-    prefetch_size -= length;
+    // Update prefetch_size as length has been updated in CopyDataToBuffer.
+    prefetch_size = length + readahead_size;
   }
 
   // Update second again if swap happened.

file/file_prefetch_buffer.h

@@ -14,6 +14,7 @@
 #include <string>
 
 #include "file/readahead_file_info.h"
+#include "monitoring/statistics.h"
 #include "port/port.h"
 #include "rocksdb/env.h"
 #include "rocksdb/file_system.h"
@@ -64,7 +65,8 @@ class FilePrefetchBuffer {
   FilePrefetchBuffer(size_t readahead_size = 0, size_t max_readahead_size = 0,
                      bool enable = true, bool track_min_offset = false,
                      bool implicit_auto_readahead = false,
-                     bool async_io = false, FileSystem* fs = nullptr)
+                     bool async_io = false, FileSystem* fs = nullptr,
+                     SystemClock* clock = nullptr, Statistics* stats = nullptr)
       : curr_(0),
         readahead_size_(readahead_size),
         initial_auto_readahead_size_(readahead_size),
@@ -80,7 +82,9 @@ class FilePrefetchBuffer {
         del_fn_(nullptr),
         async_read_in_progress_(false),
         async_io_(async_io),
-        fs_(fs) {
+        fs_(fs),
+        clock_(clock),
+        stats_(stats) {
     // If async_io_ is enabled, data is asynchronously filled in second buffer
     // while curr_ is being consumed. If data is overlapping in two buffers,
     // data is copied to third buffer to return continuous buffer.
@@ -88,12 +92,24 @@ class FilePrefetchBuffer {
   }
 
   ~FilePrefetchBuffer() {
-    // Wait for any pending async job before destroying the class object.
+    // Abort any pending async read request before destroying the class object.
     if (async_read_in_progress_ && fs_ != nullptr) {
       std::vector<void*> handles;
       handles.emplace_back(io_handle_);
-      fs_->Poll(handles, 1).PermitUncheckedError();
+      Status s = fs_->AbortIO(handles);
+      assert(s.ok());
     }
+
+    // Prefetch buffer bytes discarded.
+    uint64_t bytes_discarded = 0;
+    if (bufs_[curr_].buffer_.CurrentSize() != 0) {
+      bytes_discarded = bufs_[curr_].buffer_.CurrentSize();
+    }
+    if (bufs_[curr_ ^ 1].buffer_.CurrentSize() != 0) {
+      bytes_discarded += bufs_[curr_ ^ 1].buffer_.CurrentSize();
+    }
+    RecordInHistogram(stats_, PREFETCHED_BYTES_DISCARDED, bytes_discarded);
+
     // Release io_handle_.
     if (io_handle_ != nullptr && del_fn_ != nullptr) {
       del_fn_(io_handle_);
@@ -272,5 +288,7 @@ class FilePrefetchBuffer {
   bool async_read_in_progress_;
   bool async_io_;
   FileSystem* fs_;
+  SystemClock* clock_;
+  Statistics* stats_;
 };
 }  // namespace ROCKSDB_NAMESPACE

file/prefetch_test.cc

@@ -1288,6 +1288,10 @@ TEST_P(PrefetchTestWithPosix, ReadAsyncWithPosixFS) {
     {
       HistogramData async_read_bytes;
       options.statistics->histogramData(ASYNC_READ_BYTES, &async_read_bytes);
+      HistogramData prefetched_bytes_discarded;
+      options.statistics->histogramData(PREFETCHED_BYTES_DISCARDED,
+                                        &prefetched_bytes_discarded);
+
       // Not all platforms support iouring. In that case, ReadAsync in posix
       // won't submit async requests.
       if (read_async_called) {
@@ -1295,6 +1299,7 @@ TEST_P(PrefetchTestWithPosix, ReadAsyncWithPosixFS) {
       } else {
         ASSERT_EQ(async_read_bytes.count, 0);
       }
+      ASSERT_GT(prefetched_bytes_discarded.count, 0);
     }
   }
 

file/writable_file_writer.h

@@ -160,7 +160,9 @@ class WritableFileWriter {
   bool perform_data_verification_;
   uint32_t buffered_data_crc32c_checksum_;
   bool buffered_data_with_checksum_;
+#ifndef ROCKSDB_LITE
   Temperature temperature_;
+#endif  // ROCKSDB_LITE
 
  public:
   WritableFileWriter(
@@ -191,8 +193,10 @@ class WritableFileWriter {
         checksum_finalized_(false),
         perform_data_verification_(perform_data_verification),
         buffered_data_crc32c_checksum_(0),
-        buffered_data_with_checksum_(buffered_data_with_checksum),
-        temperature_(options.temperature) {
+        buffered_data_with_checksum_(buffered_data_with_checksum) {
+#ifndef ROCKSDB_LITE
+    temperature_ = options.temperature;
+#endif  // ROCKSDB_LITE
     assert(!use_direct_io() || max_buffer_size_ > 0);
     TEST_SYNC_POINT_CALLBACK("WritableFileWriter::WritableFileWriter:0",
                              reinterpret_cast<void*>(max_buffer_size_));

include/rocksdb/file_system.h

@@ -668,6 +668,17 @@ class FileSystem : public Customizable {
     return IOStatus::OK();
   }
 
+  // EXPERIMENTAL
+  // Abort the read IO requests submitted asynchronously. Underlying FS is
+  // required to support AbortIO API. AbortIO implementation should ensure that
+  // the all the read requests related to io_handles should be aborted and
+  // it shouldn't call the callback for these io_handles.
+  //
+  // Default implementation is to return IOStatus::OK.
+  virtual IOStatus AbortIO(std::vector<void*>& /*io_handles*/) {
+    return IOStatus::OK();
+  }
+
   // If you're adding methods here, remember to add them to EnvWrapper too.
 
  private:
@@ -1500,6 +1511,10 @@ class FileSystemWrapper : public FileSystem {
     return target_->Poll(io_handles, min_completions);
   }
 
+  virtual IOStatus AbortIO(std::vector<void*>& io_handles) override {
+    return target_->AbortIO(io_handles);
+  }
+
  protected:
   std::shared_ptr<FileSystem> target_;
 };

include/rocksdb/statistics.h

@@ -536,7 +536,12 @@ enum Histograms : uint32_t {
   // Error handler statistics
   ERROR_HANDLER_AUTORESUME_RETRY_COUNT,
 
+  // Stats related to asynchronous read requests.
   ASYNC_READ_BYTES,
+  POLL_WAIT_MICROS,
+
+  // Number of prefetched bytes discarded by RocksDB.
+  PREFETCHED_BYTES_DISCARDED,
 
   HISTOGRAM_ENUM_MAX,
 };

include/rocksdb/version.h

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

java/rocksjni/portal.h

@@ -5589,6 +5589,10 @@ class HistogramTypeJni {
         return 0x32;
       case ROCKSDB_NAMESPACE::Histograms::ASYNC_READ_BYTES:
         return 0x33;
+      case ROCKSDB_NAMESPACE::Histograms::POLL_WAIT_MICROS:
+        return 0x34;
+      case ROCKSDB_NAMESPACE::Histograms::PREFETCHED_BYTES_DISCARDED:
+        return 0x35;
       case ROCKSDB_NAMESPACE::Histograms::HISTOGRAM_ENUM_MAX:
         // 0x1F for backwards compatibility on current minor version.
         return 0x1F;
@@ -5708,6 +5712,10 @@ class HistogramTypeJni {
             ERROR_HANDLER_AUTORESUME_RETRY_COUNT;
       case 0x33:
         return ROCKSDB_NAMESPACE::Histograms::ASYNC_READ_BYTES;
+      case 0x34:
+        return ROCKSDB_NAMESPACE::Histograms::POLL_WAIT_MICROS;
+      case 0x35:
+        return ROCKSDB_NAMESPACE::Histograms::PREFETCHED_BYTES_DISCARDED;
       case 0x1F:
         // 0x1F for backwards compatibility on current minor version.
         return ROCKSDB_NAMESPACE::Histograms::HISTOGRAM_ENUM_MAX;

java/src/test/java/org/rocksdb/RocksDBTest.java

@@ -1428,7 +1428,7 @@ public class RocksDBTest {
         assertThat(livefiles.manifestFileSize).isEqualTo(59);
         assertThat(livefiles.files.size()).isEqualTo(3);
         assertThat(livefiles.files.get(0)).isEqualTo("/CURRENT");
-        assertThat(livefiles.files.get(1)).isEqualTo("/MANIFEST-000005");
+        assertThat(livefiles.files.get(1)).isEqualTo("/MANIFEST-000004");
         assertThat(livefiles.files.get(2)).isEqualTo("/OPTIONS-000007");
       }
     }

monitoring/statistics.cc

@@ -284,6 +284,9 @@ const std::vector<std::pair<Histograms, std::string>> HistogramsNameMap = {
     {ERROR_HANDLER_AUTORESUME_RETRY_COUNT,
      "rocksdb.error.handler.autoresume.retry.count"},
     {ASYNC_READ_BYTES, "rocksdb.async.read.bytes"},
+    {POLL_WAIT_MICROS, "rocksdb.poll.wait.micros"},
+    {PREFETCHED_BYTES_DISCARDED, "rocksdb.prefetched.bytes.discarded"},
+
 };
 
 std::shared_ptr<Statistics> CreateDBStatistics() {

monitoring/stats_history_test.cc

@@ -604,14 +604,10 @@ TEST_F(StatsHistoryTest, ForceManualFlushStatsCF) {
   dbfull()->TEST_WaitForStatsDumpRun(
       [&] { mock_clock_->MockSleepForSeconds(kPeriodSec); });
   // writing to all three cf, flush default cf
-  // LogNumbers: default: 16, stats: 10, pikachu: 5
-  // Since in recovery process, cfd_stats column is created after WAL is
-  // created, synced and MANIFEST is persisted, its log number which depends on
-  // logfile_number_ will be different. Since "pikachu" is never flushed, thus
-  // its log_number should be the smallest of the three.
+  // LogNumbers: default: 14, stats: 4, pikachu: 4
   ASSERT_OK(Flush());
-  ASSERT_LT(cfd_test->GetLogNumber(), cfd_stats->GetLogNumber());
-  ASSERT_LT(cfd_test->GetLogNumber(), cfd_default->GetLogNumber());
+  ASSERT_EQ(cfd_stats->GetLogNumber(), cfd_test->GetLogNumber());
+  ASSERT_LT(cfd_stats->GetLogNumber(), cfd_default->GetLogNumber());
 
   ASSERT_OK(Put("foo1", "v1"));
   ASSERT_OK(Put("bar1", "v1"));

table/block_based/block_based_table_reader.h

@@ -666,7 +666,8 @@ struct BlockBasedTable::Rep {
     fpb->reset(new FilePrefetchBuffer(
         readahead_size, max_readahead_size,
         !ioptions.allow_mmap_reads /* enable */, false /* track_min_offset */,
-        implicit_auto_readahead, async_io, ioptions.fs.get()));
+        implicit_auto_readahead, async_io, ioptions.fs.get(), ioptions.clock,
+        ioptions.stats));
   }
 
   void CreateFilePrefetchBufferIfNotExists(

table/block_based/block_based_table_reader_test.cc

@@ -5,6 +5,7 @@
 
 #include "table/block_based/block_based_table_reader.h"
 
+#include <cmath>
 #include <memory>
 #include <string>
 
@@ -330,9 +331,9 @@ class BlockBasedTableReaderCapMemoryTest
                                 CacheEntryRole::kBlockBasedTableReader>::
                                 GetDummyEntrySize() ==
                0 &&
-           cache_capacity > 2 * CacheReservationManagerImpl<
-                                    CacheEntryRole::kBlockBasedTableReader>::
-                                    GetDummyEntrySize());
+           cache_capacity >= 2 * CacheReservationManagerImpl<
+                                     CacheEntryRole::kBlockBasedTableReader>::
+                                     GetDummyEntrySize());
 
     // We need to subtract 1 for max_num_dummy_entry to account for dummy
     // entries' overhead, assumed the overhead is no greater than 1 dummy entry
@@ -347,11 +348,11 @@ class BlockBasedTableReaderCapMemoryTest
         max_num_dummy_entry *
         CacheReservationManagerImpl<
             CacheEntryRole::kBlockBasedTableReader>::GetDummyEntrySize();
-    std::size_t max_table_reader_num = static_cast<std::size_t>(
+    std::size_t max_table_reader_num_capped = static_cast<std::size_t>(
         std::floor(1.0 * cache_capacity_rounded_to_dummy_entry_multiples /
                    approx_table_reader_mem));
 
-    return max_table_reader_num;
+    return max_table_reader_num_capped;
   }
 
   void SetUp() override {
@@ -361,7 +362,7 @@ class BlockBasedTableReaderCapMemoryTest
     compression_type_ = CompressionType::kNoCompression;
 
     table_reader_res_only_cache_.reset(new BlockBasedTableReaderResOnlyCache(
-        NewLRUCache(6 * CacheReservationManagerImpl<
+        NewLRUCache(4 * CacheReservationManagerImpl<
                             CacheEntryRole::kBlockBasedTableReader>::
                             GetDummyEntrySize(),
                     0 /* num_shard_bits */, true /* strict_capacity_limit */)));
@@ -420,22 +421,44 @@ class BlockBasedTableReaderCapMemoryTest
 };
 
 INSTANTIATE_TEST_CASE_P(CapMemoryUsageUnderCacheCapacity,
-                        BlockBasedTableReaderCapMemoryTest, ::testing::Bool());
+                        BlockBasedTableReaderCapMemoryTest,
+                        ::testing::Values(true, false));
 
 TEST_P(BlockBasedTableReaderCapMemoryTest, CapMemoryUsageUnderCacheCapacity) {
-  const std::size_t max_table_reader_num = BlockBasedTableReaderCapMemoryTest::
-      CalculateMaxTableReaderNumBeforeCacheFull(
-          table_reader_res_only_cache_->GetCapacity(),
-          approx_table_reader_mem_);
+  const std::size_t max_table_reader_num_capped =
+      BlockBasedTableReaderCapMemoryTest::
+          CalculateMaxTableReaderNumBeforeCacheFull(
+              table_reader_res_only_cache_->GetCapacity(),
+              approx_table_reader_mem_);
+
+  // Acceptable estimtation errors coming from
+  // 1. overstimate max_table_reader_num_capped due to # dummy entries is high
+  // and results in metadata charge overhead greater than 1 dummy entry size
+  // (violating our assumption in calculating max_table_reader_num_capped)
+  // 2. overestimate/underestimate max_table_reader_num_capped due to the gap
+  // between ApproximateTableReaderMem() and actual table reader mem
+  std::size_t max_table_reader_num_capped_upper_bound =
+      (std::size_t)(max_table_reader_num_capped * 1.01);
+  std::size_t max_table_reader_num_capped_lower_bound =
+      (std::size_t)(max_table_reader_num_capped * 0.99);
+  std::size_t max_table_reader_num_uncapped =
+      (std::size_t)(max_table_reader_num_capped * 1.1);
+  ASSERT_GT(max_table_reader_num_uncapped,
+            max_table_reader_num_capped_upper_bound)
+      << "We need `max_table_reader_num_uncapped` > "
+         "`max_table_reader_num_capped_upper_bound` to differentiate cases "
+         "between "
+         "reserve_table_reader_memory_ == false and == true)";
 
   Status s = Status::OK();
   std::size_t opened_table_reader_num = 0;
   std::string table_name;
   std::vector<std::unique_ptr<BlockBasedTable>> tables;
-
   // Keep creating BlockBasedTableReader till hiting the memory limit based on
-  // cache capacity and creation fails or reaching a big number of table readers
-  while (s.ok() && opened_table_reader_num < 2 * max_table_reader_num) {
+  // cache capacity and creation fails (when reserve_table_reader_memory_ ==
+  // true) or reaching a specfied big number of table readers (when
+  // reserve_table_reader_memory_ == false)
+  while (s.ok() && opened_table_reader_num < max_table_reader_num_uncapped) {
     table_name = "table_" + std::to_string(opened_table_reader_num);
     CreateTable(table_name, compression_type_, kv_);
     tables.push_back(std::unique_ptr<BlockBasedTable>());
@@ -449,23 +472,14 @@ TEST_P(BlockBasedTableReaderCapMemoryTest, CapMemoryUsageUnderCacheCapacity) {
   }
 
   if (reserve_table_reader_memory_) {
-    EXPECT_TRUE(s.IsMemoryLimit() &&
-                opened_table_reader_num < 2 * max_table_reader_num)
-        << "s: " << s.ToString() << " opened_table_reader_num: "
-        << std::to_string(opened_table_reader_num);
+    EXPECT_TRUE(s.IsMemoryLimit()) << "s: " << s.ToString();
     EXPECT_TRUE(s.ToString().find("memory limit based on cache capacity") !=
                 std::string::npos);
 
-    // Acceptable estimtation errors coming from
-    // 1. overstimate max_table_reader_num due to # dummy entries is high and
-    // results in metadata charge overhead greater than 1 dummy entry size
-    // (violating our assumption in calculating max_table_reader_nums)
-    // 2. overestimate/underestimate max_table_reader_num due to the gap between
-    // ApproximateTableReaderMem() and actual table reader mem
-    EXPECT_GE(opened_table_reader_num, max_table_reader_num * 0.99);
-    EXPECT_LE(opened_table_reader_num, max_table_reader_num * 1.01);
+    EXPECT_GE(opened_table_reader_num, max_table_reader_num_capped_lower_bound);
+    EXPECT_LE(opened_table_reader_num, max_table_reader_num_capped_upper_bound);
 
-    std::size_t updated_max_table_reader_num =
+    std::size_t updated_max_table_reader_num_capped =
         BlockBasedTableReaderCapMemoryTest::
             CalculateMaxTableReaderNumBeforeCacheFull(
                 table_reader_res_only_cache_->GetCapacity() / 2,
@@ -473,7 +487,7 @@ TEST_P(BlockBasedTableReaderCapMemoryTest, CapMemoryUsageUnderCacheCapacity) {
 
     // Keep deleting BlockBasedTableReader to lower down memory usage from the
     // memory limit to make the next creation succeeds
-    while (opened_table_reader_num >= updated_max_table_reader_num) {
+    while (opened_table_reader_num >= updated_max_table_reader_num_capped) {
       tables.pop_back();
       --opened_table_reader_num;
     }
@@ -489,7 +503,8 @@ TEST_P(BlockBasedTableReaderCapMemoryTest, CapMemoryUsageUnderCacheCapacity) {
     tables.clear();
     EXPECT_EQ(table_reader_res_only_cache_->GetPinnedUsage(), 0);
   } else {
-    EXPECT_TRUE(s.ok() && opened_table_reader_num == 2 * max_table_reader_num)
+    EXPECT_TRUE(s.ok() &&
+                opened_table_reader_num == max_table_reader_num_uncapped)
         << "s: " << s.ToString() << " opened_table_reader_num: "
         << std::to_string(opened_table_reader_num);
     EXPECT_EQ(table_reader_res_only_cache_->GetPinnedUsage(), 0);