Update version.h and HISTORY.md for 7.1.1

Summary:
Update stats in random_access_file_reader for Read and
ReadAsync API to take into account the read latency for async
prefetching.

It also fixes ERROR_HANDLER_AUTORESUME_RETRY_COUNT stat whose value was
incorrect in portal.h

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

Test Plan: Update unit test

Reviewed By: anand1976

Differential Revision: D35433081

Pulled By: akankshamahajan15

fbshipit-source-id: aeec3901270e58a003ce6b5214bd25ddcb3a12a9

HISTORY.md

@@ -1,8 +1,9 @@
 # Rocksdb Change Log
-## 7.1.0 (03/21/2022)
-### Public API changes
-* Add DB::OpenAndTrimHistory API. This API will open DB and trim data to the timestamp specified by trim_ts (The data with timestamp larger than specified trim bound will be removed). This API should only be used at a timestamp-enabled column families recovery. If the column family doesn't have timestamp enabled, this API won't trim any data on that column family. This API is not compatible with avoid_flush_during_recovery option.
+## 7.1.1 (04/07/2022)
+### Bug Fixes
+* Fix segfault in FilePrefetchBuffer with async_io as it doesn't wait for pending jobs to complete on destruction.
 
+## 7.1.0 (03/23/2022)
 ### New Features
 * Allow WriteBatchWithIndex to index a WriteBatch that includes keys with user-defined timestamps. The index itself does not have timestamp.
 * Add support for user-defined timestamps to write-committed transaction without API change. The `TransactionDB` layer APIs do not allow timestamps because we require that all user-defined-timestamps-aware operations go through the `Transaction` APIs.
@@ -13,10 +14,12 @@
 * Experimental support for async_io in ReadOptions which is used by FilePrefetchBuffer to prefetch some of the data asynchronously,  if reads are sequential and auto readahead is enabled by rocksdb internally.
 
 ### Bug Fixes
+* Fixed a major performance bug in which Bloom filters generated by pre-7.0 releases are not read by early 7.0.x releases (and vice-versa) due to changes to FilterPolicy::Name() in #9590. This can severely impact read performance and read I/O on upgrade or downgrade with existing DB, but not data correctness.
 * Fixed a data race on `versions_` between `DBImpl::ResumeImpl()` and threads waiting for recovery to complete (#9496)
 * Fixed a bug caused by race among flush, incoming writes and taking snapshots. Queries to snapshots created with these race condition can return incorrect result, e.g. resurfacing deleted data.
 * Fixed a bug that DB flush uses `options.compression` even `options.compression_per_level` is set.
 * Fixed a bug that DisableManualCompaction may assert when disable an unscheduled manual compaction.
+* Fix a race condition when cancel manual compaction with `DisableManualCompaction`. Also DB close can cancel the manual compaction thread.
 * Fixed a potential timer crash when open close DB concurrently.
 * Fixed a race condition for `alive_log_files_` in non-two-write-queues mode. The race is between the write_thread_ in WriteToWAL() and another thread executing `FindObsoleteFiles()`. The race condition will be caught if `__glibcxx_requires_nonempty` is enabled.
 * Fixed a bug that `Iterator::Refresh()` reads stale keys after DeleteRange() performed.
@@ -25,12 +28,11 @@
 * Fixed a race condition when mmaping a WritableFile on POSIX.
 
 ### Public API changes
-* Remove BlockBasedTableOptions.hash_index_allow_collision which already takes no effect.
+* Added pure virtual FilterPolicy::CompatibilityName(), which is needed for fixing major performance bug involving FilterPolicy naming in SST metadata without affecting Customizable aspect of FilterPolicy. This change only affects those with their own custom or wrapper FilterPolicy classes.
 * `options.compression_per_level` is dynamically changeable with `SetOptions()`.
 * Added `WriteOptions::rate_limiter_priority`. When set to something other than `Env::IO_TOTAL`, the internal rate limiter (`DBOptions::rate_limiter`) will be charged at the specified priority for writes associated with the API to which the `WriteOptions` was provided. Currently the support covers automatic WAL flushes, which happen during live updates (`Put()`, `Write()`, `Delete()`, etc.) when `WriteOptions::disableWAL == false` and `DBOptions::manual_wal_flush == false`.
-
-### Bug Fixes
-* Fix a race condition when cancel manual compaction with `DisableManualCompaction`. Also DB close can cancel the manual compaction thread.
+* Add DB::OpenAndTrimHistory API. This API will open DB and trim data to the timestamp specified by trim_ts (The data with timestamp larger than specified trim bound will be removed). This API should only be used at a timestamp-enabled column families recovery. If the column family doesn't have timestamp enabled, this API won't trim any data on that column family. This API is not compatible with avoid_flush_during_recovery option.
+* Remove BlockBasedTableOptions.hash_index_allow_collision which already takes no effect.
 
 ## 7.0.0 (02/20/2022)
 ### Bug Fixes

Makefile

@@ -2046,7 +2046,7 @@ SHA256_CMD = sha256sum
 
 ZLIB_VER ?= 1.2.11
 ZLIB_SHA256 ?= c3e5e9fdd5004dcb542feda5ee4f0ff0744628baf8ed2dd5d66f8ca1197cb1a1
-ZLIB_DOWNLOAD_BASE ?= http://zlib.net
+ZLIB_DOWNLOAD_BASE ?= https://zlib.net/fossils
 BZIP2_VER ?= 1.0.8
 BZIP2_SHA256 ?= ab5a03176ee106d3f0fa90e381da478ddae405918153cca248e682cd0c4a2269
 BZIP2_DOWNLOAD_BASE ?= http://sourceware.org/pub/bzip2

db/c.cc

@@ -3752,6 +3752,9 @@ rocksdb_filterpolicy_t* rocksdb_filterpolicy_create_bloom_format(
     const FilterPolicy* rep_;
     ~Wrapper() override { delete rep_; }
     const char* Name() const override { return rep_->Name(); }
+    const char* CompatibilityName() const override {
+      return rep_->CompatibilityName();
+    }
     // No need to override GetFilterBitsBuilder if this one is overridden
     ROCKSDB_NAMESPACE::FilterBitsBuilder* GetBuilderWithContext(
         const ROCKSDB_NAMESPACE::FilterBuildingContext& context)
@@ -3789,6 +3792,9 @@ rocksdb_filterpolicy_t* rocksdb_filterpolicy_create_ribbon_format(
     const FilterPolicy* rep_;
     ~Wrapper() override { delete rep_; }
     const char* Name() const override { return rep_->Name(); }
+    const char* CompatibilityName() const override {
+      return rep_->CompatibilityName();
+    }
     ROCKSDB_NAMESPACE::FilterBitsBuilder* GetBuilderWithContext(
         const ROCKSDB_NAMESPACE::FilterBuildingContext& context)
         const override {

db/db_bloom_filter_test.cc

@@ -1638,9 +1638,15 @@ class LevelAndStyleCustomFilterPolicy : public FilterPolicy {
         policy_l0_other_(NewBloomFilterPolicy(bpk_l0_other)),
         policy_otherwise_(NewBloomFilterPolicy(bpk_otherwise)) {}
 
+  const char* Name() const override {
+    return "LevelAndStyleCustomFilterPolicy";
+  }
+
   // OK to use built-in policy name because we are deferring to a
   // built-in builder. We aren't changing the serialized format.
-  const char* Name() const override { return policy_fifo_->Name(); }
+  const char* CompatibilityName() const override {
+    return policy_fifo_->CompatibilityName();
+  }
 
   FilterBitsBuilder* GetBuilderWithContext(
       const FilterBuildingContext& context) const override {

env/env_test.cc

@@ -85,6 +85,8 @@ struct Deleter {
   void (*fn_)(void*);
 };
 
+extern "C" bool RocksDbIOUringEnable() { return true; }
+
 std::unique_ptr<char, Deleter> NewAligned(const size_t size, const char ch) {
   char* ptr = nullptr;
 #ifdef OS_WIN
@@ -1256,7 +1258,7 @@ TEST_P(EnvPosixTestWithParam, MultiRead) {
     // Random Read
     Random rnd(301 + attempt);
     ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack(
-        "UpdateResults:io_uring_result", [&](void* arg) {
+        "UpdateResults::io_uring_result", [&](void* arg) {
           if (attempt > 0) {
             // No failure in the first attempt.
             size_t& bytes_read = *static_cast<size_t*>(arg);
@@ -1326,7 +1328,7 @@ TEST_F(EnvPosixTest, MultiReadNonAlignedLargeNum) {
     const int num_reads = rnd.Uniform(512) + 1;
 
     ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack(
-        "UpdateResults:io_uring_result", [&](void* arg) {
+        "UpdateResults::io_uring_result", [&](void* arg) {
           if (attempt > 5) {
             // Improve partial result rates in second half of the run to
             // cover the case of repeated partial results.
@@ -3308,7 +3310,6 @@ TEST_F(TestAsyncRead, ReadAsync) {
     }
   }
 }
-
 }  // namespace ROCKSDB_NAMESPACE
 
 int main(int argc, char** argv) {

env/fs_posix.cc

@@ -1045,9 +1045,12 @@ class PosixFileSystem : public FileSystem {
 
   // EXPERIMENTAL
   //
-  // TODO akankshamahajan: Update Poll API to take into account min_completions
+  // TODO akankshamahajan:
+  // 1. Update Poll API to take into account min_completions
   // and returns if number of handles in io_handles (any order) completed is
   // equal to atleast min_completions.
+  // 2. Currently in case of direct_io, Read API is called because of which call
+  // to Poll API fails as it expects IOHandle to be populated.
   virtual IOStatus Poll(std::vector<void*>& io_handles,
                         size_t /*min_completions*/) override {
 #if defined(ROCKSDB_IOURING_PRESENT)
@@ -1094,12 +1097,14 @@ class PosixFileSystem : public FileSystem {
         req.offset = posix_handle->offset;
         req.len = posix_handle->len;
         size_t finished_len = 0;
+        size_t bytes_read = 0;
         UpdateResult(cqe, "", req.len, posix_handle->iov.iov_len,
-                     true /*async_read*/, finished_len, &req);
+                     true /*async_read*/, finished_len, &req, bytes_read);
         posix_handle->is_finished = true;
         io_uring_cqe_seen(iu, cqe);
         posix_handle->cb(req, posix_handle->cb_arg);
         (void)finished_len;
+        (void)bytes_read;
 
         if (static_cast<Posix_IOHandle*>(io_handles[i]) == posix_handle) {
           break;

env/io_posix.cc

@@ -744,31 +744,36 @@ IOStatus PosixRandomAccessFile::MultiRead(FSReadRequest* reqs,
       wrap_cache.erase(wrap_check);
 
       FSReadRequest* req = req_wrap->req;
+      size_t bytes_read = 0;
       UpdateResult(cqe, filename_, req->len, req_wrap->iov.iov_len,
-                   false /*async_read*/, req_wrap->finished_len, req);
+                   false /*async_read*/, req_wrap->finished_len, req,
+                   bytes_read);
       int32_t res = cqe->res;
-      if (res == 0) {
-        /// cqe->res == 0 can means EOF, or can mean partial results. See
-        // comment
-        // https://github.com/facebook/rocksdb/pull/6441#issuecomment-589843435
-        // Fall back to pread in this case.
-        if (use_direct_io() && !IsSectorAligned(req_wrap->finished_len,
-                                                GetRequiredBufferAlignment())) {
-          // Bytes reads don't fill sectors. Should only happen at the end
-          // of the file.
-          req->result = Slice(req->scratch, req_wrap->finished_len);
-          req->status = IOStatus::OK();
-        } else {
-          Slice tmp_slice;
-          req->status =
-              Read(req->offset + req_wrap->finished_len,
-                   req->len - req_wrap->finished_len, options, &tmp_slice,
-                   req->scratch + req_wrap->finished_len, dbg);
-          req->result =
-              Slice(req->scratch, req_wrap->finished_len + tmp_slice.size());
+      if (res >= 0) {
+        if (bytes_read == 0) {
+          /// cqe->res == 0 can means EOF, or can mean partial results. See
+          // comment
+          // https://github.com/facebook/rocksdb/pull/6441#issuecomment-589843435
+          // Fall back to pread in this case.
+          if (use_direct_io() &&
+              !IsSectorAligned(req_wrap->finished_len,
+                               GetRequiredBufferAlignment())) {
+            // Bytes reads don't fill sectors. Should only happen at the end
+            // of the file.
+            req->result = Slice(req->scratch, req_wrap->finished_len);
+            req->status = IOStatus::OK();
+          } else {
+            Slice tmp_slice;
+            req->status =
+                Read(req->offset + req_wrap->finished_len,
+                     req->len - req_wrap->finished_len, options, &tmp_slice,
+                     req->scratch + req_wrap->finished_len, dbg);
+            req->result =
+                Slice(req->scratch, req_wrap->finished_len + tmp_slice.size());
+          }
+        } else if (bytes_read < req_wrap->iov.iov_len) {
+          incomplete_rq_list.push_back(req_wrap);
         }
-      } else if (res > 0 && res < static_cast<int32_t>(req_wrap->iov.iov_len)) {
-        incomplete_rq_list.push_back(req_wrap);
       }
       io_uring_cqe_seen(iu, cqe);
     }
@@ -896,8 +901,8 @@ IOStatus PosixRandomAccessFile::ReadAsync(
 
   // Initialize Posix_IOHandle.
   posix_handle->iu = iu;
-  posix_handle->iov.iov_base = posix_handle->scratch;
-  posix_handle->iov.iov_len = posix_handle->len;
+  posix_handle->iov.iov_base = req.scratch;
+  posix_handle->iov.iov_len = req.len;
   posix_handle->cb = cb;
   posix_handle->cb_arg = cb_arg;
   posix_handle->offset = req.offset;

env/io_posix.h

@@ -66,12 +66,13 @@ struct Posix_IOHandle {
 
 inline void UpdateResult(struct io_uring_cqe* cqe, const std::string& file_name,
                          size_t len, size_t iov_len, bool async_read,
-                         size_t& finished_len, FSReadRequest* req) {
+                         size_t& finished_len, FSReadRequest* req,
+                         size_t& bytes_read) {
   if (cqe->res < 0) {
     req->result = Slice(req->scratch, 0);
     req->status = IOError("Req failed", file_name, cqe->res);
   } else {
-    size_t bytes_read = static_cast<size_t>(cqe->res);
+    bytes_read = static_cast<size_t>(cqe->res);
     TEST_SYNC_POINT_CALLBACK("UpdateResults::io_uring_result", &bytes_read);
     if (bytes_read == iov_len) {
       req->result = Slice(req->scratch, req->len);

file/file_prefetch_buffer.cc

@@ -66,6 +66,17 @@ void FilePrefetchBuffer::CalculateOffsetAndLen(size_t alignment,
     // chunk_len is greater than 0.
     bufs_[index].buffer_.RefitTail(static_cast<size_t>(chunk_offset_in_buffer),
                                    static_cast<size_t>(chunk_len));
+  } else if (chunk_len > 0) {
+    // For async prefetching, it doesn't call RefitTail with chunk_len > 0.
+    // Allocate new buffer if needed because aligned buffer calculate remaining
+    // buffer as capacity_ - cursize_ which might not be the case in this as we
+    // are not refitting.
+    // TODO akanksha: Update the condition when asynchronous prefetching is
+    // stable.
+    bufs_[index].buffer_.Alignment(alignment);
+    bufs_[index].buffer_.AllocateNewBuffer(
+        static_cast<size_t>(roundup_len), copy_data_to_new_buffer,
+        chunk_offset_in_buffer, static_cast<size_t>(chunk_len));
   }
 }
 
@@ -100,13 +111,6 @@ Status FilePrefetchBuffer::ReadAsync(const IOOptions& opts,
                                      Env::IOPriority rate_limiter_priority,
                                      uint64_t read_len, uint64_t chunk_len,
                                      uint64_t rounddown_start, uint32_t index) {
-  // Reset io_handle.
-  if (io_handle_ != nullptr && del_fn_ != nullptr) {
-    del_fn_(io_handle_);
-    io_handle_ = nullptr;
-    del_fn_ = nullptr;
-  }
-
   // callback for async read request.
   auto fp = std::bind(&FilePrefetchBuffer::PrefetchAsyncCallback, this,
                       std::placeholders::_1, std::placeholders::_2);
@@ -118,6 +122,7 @@ Status FilePrefetchBuffer::ReadAsync(const IOOptions& opts,
   req.scratch = bufs_[index].buffer_.BufferStart() + chunk_len;
   Status s = reader->ReadAsync(req, opts, fp, nullptr /*cb_arg*/, &io_handle_,
                                &del_fn_, rate_limiter_priority);
+  req.status.PermitUncheckedError();
   if (s.ok()) {
     async_read_in_progress_ = true;
   }
@@ -210,24 +215,34 @@ void FilePrefetchBuffer::CopyDataToBuffer(uint32_t src, uint64_t& offset,
 //        the asynchronous prefetching in second buffer.
 Status FilePrefetchBuffer::PrefetchAsync(const IOOptions& opts,
                                          RandomAccessFileReader* reader,
-                                         FileSystem* fs, uint64_t offset,
-                                         size_t length, size_t readahead_size,
+                                         uint64_t offset, size_t length,
+                                         size_t readahead_size,
                                          Env::IOPriority rate_limiter_priority,
                                          bool& copy_to_third_buffer) {
   if (!enable_) {
     return Status::OK();
   }
-  if (async_read_in_progress_ && fs != nullptr) {
+  if (async_read_in_progress_ && fs_ != nullptr) {
     // Wait for prefetch data to complete.
     // No mutex is needed as PrefetchAsyncCallback updates the result in second
     // buffer and FilePrefetchBuffer should wait for Poll before accessing the
     // second buffer.
     std::vector<void*> handles;
     handles.emplace_back(io_handle_);
-    fs->Poll(handles, 1).PermitUncheckedError();
+    fs_->Poll(handles, 1).PermitUncheckedError();
   }
 
-  // TODO akanksha: Update TEST_SYNC_POINT after new tests are added.
+  // Reset and Release io_handle_ after the Poll API as request has been
+  // completed.
+  async_read_in_progress_ = false;
+  if (io_handle_ != nullptr && del_fn_ != nullptr) {
+    del_fn_(io_handle_);
+    io_handle_ = nullptr;
+    del_fn_ = nullptr;
+  }
+
+  // TODO akanksha: Update TEST_SYNC_POINT after Async APIs are merged with
+  // normal prefetching.
   TEST_SYNC_POINT("FilePrefetchBuffer::Prefetch:Start");
   Status s;
   size_t prefetch_size = length + readahead_size;
@@ -236,34 +251,47 @@ Status FilePrefetchBuffer::PrefetchAsync(const IOOptions& opts,
   // Index of second buffer.
   uint32_t second = curr_ ^ 1;
 
+  // First clear the buffers if it contains outdated data. Outdated data can be
+  // because previous sequential reads were read from the cache instead of these
+  // buffer.
+  {
+    if (bufs_[curr_].buffer_.CurrentSize() > 0 &&
+        offset >= bufs_[curr_].offset_ + bufs_[curr_].buffer_.CurrentSize()) {
+      bufs_[curr_].buffer_.Clear();
+    }
+    if (bufs_[second].buffer_.CurrentSize() > 0 &&
+        offset >= bufs_[second].offset_ + bufs_[second].buffer_.CurrentSize()) {
+      bufs_[second].buffer_.Clear();
+    }
+  }
+
   // If data is in second buffer, make it curr_. Second buffer can be either
   // partial filled or full.
   if (bufs_[second].buffer_.CurrentSize() > 0 &&
       offset >= bufs_[second].offset_ &&
-      offset <= bufs_[second].offset_ + bufs_[second].buffer_.CurrentSize()) {
+      offset < bufs_[second].offset_ + bufs_[second].buffer_.CurrentSize()) {
     // Clear the curr_ as buffers have been swapped and curr_ contains the
-    // outdated data.
+    // outdated data and switch the buffers.
     bufs_[curr_].buffer_.Clear();
-    // Switch the buffers.
     curr_ = curr_ ^ 1;
     second = curr_ ^ 1;
   }
-
-  // If second buffer contains outdated data, clear it for async prefetching.
-  // Outdated can be because previous sequential reads were read from the cache
-  // instead of this buffer.
-  if (bufs_[second].buffer_.CurrentSize() > 0 &&
-      offset >= bufs_[second].offset_ + bufs_[second].buffer_.CurrentSize()) {
-    bufs_[second].buffer_.Clear();
+  // After swap check if all the requested bytes are in curr_, it will go for
+  // async prefetching only.
+  if (bufs_[curr_].buffer_.CurrentSize() > 0 &&
+      offset + length <=
+          bufs_[curr_].offset_ + bufs_[curr_].buffer_.CurrentSize()) {
+    offset += length;
+    length = 0;
+    prefetch_size -= length;
   }
-
   // Data is overlapping i.e. some of the data is in curr_ buffer and remaining
   // in second buffer.
   if (bufs_[curr_].buffer_.CurrentSize() > 0 &&
       bufs_[second].buffer_.CurrentSize() > 0 &&
       offset >= bufs_[curr_].offset_ &&
       offset < bufs_[curr_].offset_ + bufs_[curr_].buffer_.CurrentSize() &&
-      offset + prefetch_size > bufs_[second].offset_) {
+      offset + length > bufs_[second].offset_) {
     // Allocate new buffer to third buffer;
     bufs_[2].buffer_.Clear();
     bufs_[2].buffer_.Alignment(alignment);
@@ -273,12 +301,10 @@ Status FilePrefetchBuffer::PrefetchAsync(const IOOptions& opts,
 
     // Move data from curr_ buffer to third.
     CopyDataToBuffer(curr_, offset, length);
-
     if (length == 0) {
       // Requested data has been copied and curr_ still has unconsumed data.
       return s;
     }
-
     CopyDataToBuffer(second, offset, length);
     // Length == 0: All the requested data has been copied to third buffer. It
     // should go for only async prefetching.
@@ -306,6 +332,7 @@ Status FilePrefetchBuffer::PrefetchAsync(const IOOptions& opts,
   if (length > 0) {
     CalculateOffsetAndLen(alignment, offset, roundup_len1, curr_,
                           false /*refit_tail*/, chunk_len1);
+    assert(roundup_len1 >= chunk_len1);
     read_len1 = static_cast<size_t>(roundup_len1 - chunk_len1);
   }
   {
@@ -316,7 +343,7 @@ Status FilePrefetchBuffer::PrefetchAsync(const IOOptions& opts,
         Roundup(rounddown_start2 + readahead_size, alignment);
 
     // For length == 0, do the asynchronous prefetching in second instead of
-    // synchronous prefetching of remaining prefetch_size.
+    // synchronous prefetching in curr_.
     if (length == 0) {
       rounddown_start2 =
           bufs_[curr_].offset_ + bufs_[curr_].buffer_.CurrentSize();
@@ -330,8 +357,8 @@ Status FilePrefetchBuffer::PrefetchAsync(const IOOptions& opts,
 
     // Update the buffer offset.
     bufs_[second].offset_ = rounddown_start2;
+    assert(roundup_len2 >= chunk_len2);
     uint64_t read_len2 = static_cast<size_t>(roundup_len2 - chunk_len2);
-
     ReadAsync(opts, reader, rate_limiter_priority, read_len2, chunk_len2,
               rounddown_start2, second)
         .PermitUncheckedError();
@@ -344,7 +371,6 @@ Status FilePrefetchBuffer::PrefetchAsync(const IOOptions& opts,
       return s;
     }
   }
-
   // Copy remaining requested bytes to third_buffer.
   if (copy_to_third_buffer && length > 0) {
     CopyDataToBuffer(curr_, offset, length);
@@ -416,8 +442,8 @@ bool FilePrefetchBuffer::TryReadFromCache(const IOOptions& opts,
 bool FilePrefetchBuffer::TryReadFromCacheAsync(
     const IOOptions& opts, RandomAccessFileReader* reader, uint64_t offset,
     size_t n, Slice* result, Status* status,
-    Env::IOPriority rate_limiter_priority, bool for_compaction /* = false */,
-    FileSystem* fs) {
+    Env::IOPriority rate_limiter_priority, bool for_compaction /* = false */
+) {
   if (track_min_offset_ && offset < min_offset_read_) {
     min_offset_read_ = static_cast<size_t>(offset);
   }
@@ -452,7 +478,7 @@ bool FilePrefetchBuffer::TryReadFromCacheAsync(
         if (implicit_auto_readahead_ && async_io_) {
           // Prefetch n + readahead_size_/2 synchronously as remaining
           // readahead_size_/2 will be prefetched asynchronously.
-          s = PrefetchAsync(opts, reader, fs, offset, n, readahead_size_ / 2,
+          s = PrefetchAsync(opts, reader, offset, n, readahead_size_ / 2,
                             rate_limiter_priority, copy_to_third_buffer);
         } else {
           s = Prefetch(opts, reader, offset, n + readahead_size_,
@@ -489,7 +515,6 @@ bool FilePrefetchBuffer::TryReadFromCacheAsync(
 
 void FilePrefetchBuffer::PrefetchAsyncCallback(const FSReadRequest& req,
                                                void* /*cb_arg*/) {
-  async_read_in_progress_ = false;
   uint32_t index = curr_ ^ 1;
   if (req.status.ok()) {
     if (req.offset + req.result.size() <=
@@ -505,12 +530,5 @@ void FilePrefetchBuffer::PrefetchAsyncCallback(const FSReadRequest& req,
     size_t current_size = bufs_[index].buffer_.CurrentSize();
     bufs_[index].buffer_.Size(current_size + req.result.size());
   }
-
-  // Release io_handle_.
-  if (io_handle_ != nullptr && del_fn_ != nullptr) {
-    del_fn_(io_handle_);
-    io_handle_ = nullptr;
-    del_fn_ = nullptr;
-  }
 }
 }  // namespace ROCKSDB_NAMESPACE

file/file_prefetch_buffer.h

@@ -65,7 +65,7 @@ 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)
+                     bool async_io = false, FileSystem* fs = nullptr)
       : curr_(0),
         readahead_size_(readahead_size),
         max_readahead_size_(max_readahead_size),
@@ -79,13 +79,29 @@ class FilePrefetchBuffer {
         io_handle_(nullptr),
         del_fn_(nullptr),
         async_read_in_progress_(false),
-        async_io_(async_io) {
+        async_io_(async_io),
+        fs_(fs) {
     // 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.
     bufs_.resize(3);
   }
 
+  ~FilePrefetchBuffer() {
+    // Wait for any pending async job 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();
+    }
+    // Release io_handle_.
+    if (io_handle_ != nullptr && del_fn_ != nullptr) {
+      del_fn_(io_handle_);
+      io_handle_ = nullptr;
+      del_fn_ = nullptr;
+    }
+  }
+
   // Load data into the buffer from a file.
   // reader                : the file reader.
   // offset                : the file offset to start reading from.
@@ -100,8 +116,7 @@ class FilePrefetchBuffer {
                   Env::IOPriority rate_limiter_priority);
 
   Status PrefetchAsync(const IOOptions& opts, RandomAccessFileReader* reader,
-                       FileSystem* fs, uint64_t offset, size_t length,
-                       size_t readahead_size,
+                       uint64_t offset, size_t length, size_t readahead_size,
                        Env::IOPriority rate_limiter_priority,
                        bool& copy_to_third_buffer);
 
@@ -129,7 +144,7 @@ class FilePrefetchBuffer {
                              RandomAccessFileReader* reader, uint64_t offset,
                              size_t n, Slice* result, Status* status,
                              Env::IOPriority rate_limiter_priority,
-                             bool for_compaction /* = false */, FileSystem* fs);
+                             bool for_compaction /* = false */);
 
   // The minimum `offset` ever passed to TryReadFromCache(). This will nly be
   // tracked if track_min_offset = true.
@@ -256,5 +271,6 @@ class FilePrefetchBuffer {
   IOHandleDeleter del_fn_;
   bool async_read_in_progress_;
   bool async_io_;
+  FileSystem* fs_;
 };
 }  // namespace ROCKSDB_NAMESPACE

file/prefetch_test.cc

@@ -694,8 +694,10 @@ TEST_P(PrefetchTest1, DBIterLevelReadAhead) {
   options.write_buffer_size = 1024;
   options.create_if_missing = true;
   options.compression = kNoCompression;
+  options.statistics = CreateDBStatistics();
   options.env = env.get();
-  if (std::get<0>(GetParam())) {
+  bool use_direct_io = std::get<0>(GetParam());
+  if (use_direct_io) {
     options.use_direct_reads = true;
     options.use_direct_io_for_flush_and_compaction = true;
   }
@@ -708,8 +710,7 @@ TEST_P(PrefetchTest1, DBIterLevelReadAhead) {
   options.table_factory.reset(NewBlockBasedTableFactory(table_options));
 
   Status s = TryReopen(options);
-  if (std::get<0>(GetParam()) &&
-      (s.IsNotSupported() || s.IsInvalidArgument())) {
+  if (use_direct_io && (s.IsNotSupported() || s.IsInvalidArgument())) {
     // If direct IO is not supported, skip the test
     return;
   } else {
@@ -766,6 +767,8 @@ TEST_P(PrefetchTest1, DBIterLevelReadAhead) {
       // TODO akanksha: Remove after adding new units.
       ro.async_io = true;
     }
+
+    ASSERT_OK(options.statistics->Reset());
     auto iter = std::unique_ptr<Iterator>(db_->NewIterator(ro));
     int num_keys = 0;
     for (iter->SeekToFirst(); iter->Valid(); iter->Next()) {
@@ -773,15 +776,25 @@ TEST_P(PrefetchTest1, DBIterLevelReadAhead) {
       num_keys++;
     }
     ASSERT_EQ(num_keys, total_keys);
-
     ASSERT_GT(buff_prefetch_count, 0);
-    buff_prefetch_count = 0;
     // For index and data blocks.
     if (is_adaptive_readahead) {
       ASSERT_EQ(readahead_carry_over_count, 2 * (num_sst_files - 1));
     } else {
       ASSERT_EQ(readahead_carry_over_count, 0);
     }
+
+    // Check stats to make sure async prefetch is done.
+    {
+      HistogramData async_read_bytes;
+      options.statistics->histogramData(ASYNC_READ_BYTES, &async_read_bytes);
+      if (ro.async_io && !use_direct_io) {
+        ASSERT_GT(async_read_bytes.count, 0);
+      } else {
+        ASSERT_EQ(async_read_bytes.count, 0);
+      }
+    }
+
     SyncPoint::GetInstance()->DisableProcessing();
     SyncPoint::GetInstance()->ClearAllCallBacks();
   }
@@ -902,6 +915,8 @@ TEST_P(PrefetchTest2, DecreaseReadAheadIfInCache) {
     options.use_direct_reads = true;
     options.use_direct_io_for_flush_and_compaction = true;
   }
+
+  options.statistics = CreateDBStatistics();
   BlockBasedTableOptions table_options;
   std::shared_ptr<Cache> cache = NewLRUCache(4 * 1024 * 1024, 2);  // 8MB
   table_options.block_cache = cache;
@@ -948,7 +963,6 @@ TEST_P(PrefetchTest2, DecreaseReadAheadIfInCache) {
   SyncPoint::GetInstance()->EnableProcessing();
   ReadOptions ro;
   ro.adaptive_readahead = true;
-  // TODO akanksha: Remove after adding new units.
   ro.async_io = true;
   {
     /*
@@ -964,7 +978,9 @@ TEST_P(PrefetchTest2, DecreaseReadAheadIfInCache) {
     iter->Seek(BuildKey(1019));
     buff_prefetch_count = 0;
   }
+
   {
+    ASSERT_OK(options.statistics->Reset());
     // After caching, blocks will be read from cache (Sequential blocks)
     auto iter = std::unique_ptr<Iterator>(db_->NewIterator(ro));
     iter->Seek(BuildKey(0));
@@ -1008,11 +1024,118 @@ TEST_P(PrefetchTest2, DecreaseReadAheadIfInCache) {
     ASSERT_TRUE(iter->Valid());
     ASSERT_EQ(current_readahead_size, expected_current_readahead_size);
     ASSERT_EQ(buff_prefetch_count, 2);
+
+    // Check stats to make sure async prefetch is done.
+    {
+      HistogramData async_read_bytes;
+      options.statistics->histogramData(ASYNC_READ_BYTES, &async_read_bytes);
+      if (GetParam()) {
+        ASSERT_EQ(async_read_bytes.count, 0);
+      } else {
+        ASSERT_GT(async_read_bytes.count, 0);
+      }
+    }
+
     buff_prefetch_count = 0;
   }
   Close();
 }
 
+extern "C" bool RocksDbIOUringEnable() { return true; }
+
+// Tests the default implementation of ReadAsync API with PosixFileSystem.
+TEST_F(PrefetchTest2, ReadAsyncWithPosixFS) {
+  if (mem_env_ || encrypted_env_) {
+    ROCKSDB_GTEST_SKIP("Test requires non-mem or non-encrypted environment");
+    return;
+  }
+
+  const int kNumKeys = 1000;
+  std::shared_ptr<MockFS> fs = std::make_shared<MockFS>(
+      FileSystem::Default(), /*support_prefetch=*/false);
+  std::unique_ptr<Env> env(new CompositeEnvWrapper(env_, fs));
+
+  bool use_direct_io = false;
+  Options options = CurrentOptions();
+  options.write_buffer_size = 1024;
+  options.create_if_missing = true;
+  options.compression = kNoCompression;
+  options.env = env.get();
+  options.statistics = CreateDBStatistics();
+  if (use_direct_io) {
+    options.use_direct_reads = true;
+    options.use_direct_io_for_flush_and_compaction = true;
+  }
+  BlockBasedTableOptions table_options;
+  table_options.no_block_cache = true;
+  table_options.cache_index_and_filter_blocks = false;
+  table_options.metadata_block_size = 1024;
+  table_options.index_type =
+      BlockBasedTableOptions::IndexType::kTwoLevelIndexSearch;
+  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);
+  }
+
+  int total_keys = 0;
+  // Write the keys.
+  {
+    WriteBatch batch;
+    Random rnd(309);
+    for (int j = 0; j < 5; j++) {
+      for (int i = j * kNumKeys; i < (j + 1) * kNumKeys; i++) {
+        ASSERT_OK(batch.Put(BuildKey(i), rnd.RandomString(1000)));
+        total_keys++;
+      }
+      ASSERT_OK(db_->Write(WriteOptions(), &batch));
+      ASSERT_OK(Flush());
+    }
+    MoveFilesToLevel(2);
+  }
+
+  int buff_prefetch_count = 0;
+  SyncPoint::GetInstance()->SetCallBack("FilePrefetchBuffer::Prefetch:Start",
+                                        [&](void*) { buff_prefetch_count++; });
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->EnableProcessing();
+
+  // Read the keys.
+  {
+    ReadOptions ro;
+    ro.adaptive_readahead = true;
+    ro.async_io = true;
+
+    ASSERT_OK(options.statistics->Reset());
+    auto iter = std::unique_ptr<Iterator>(db_->NewIterator(ro));
+    int num_keys = 0;
+    for (iter->SeekToFirst(); iter->Valid(); iter->Next()) {
+      ASSERT_OK(iter->status());
+      num_keys++;
+    }
+    ASSERT_EQ(num_keys, total_keys);
+    ASSERT_GT(buff_prefetch_count, 0);
+
+    // Check stats to make sure async prefetch is done.
+    {
+      HistogramData async_read_bytes;
+      options.statistics->histogramData(ASYNC_READ_BYTES, &async_read_bytes);
+#if defined(ROCKSDB_IOURING_PRESENT)
+      ASSERT_GT(async_read_bytes.count, 0);
+#else
+      ASSERT_EQ(async_read_bytes.count, 0);
+#endif
+    }
+  }
+
+  SyncPoint::GetInstance()->DisableProcessing();
+  SyncPoint::GetInstance()->ClearAllCallBacks();
+
+  Close();
+}
 }  // namespace ROCKSDB_NAMESPACE
 
 int main(int argc, char** argv) {

file/random_access_file_reader.cc

@@ -425,19 +425,75 @@ IOStatus RandomAccessFileReader::PrepareIOOptions(const ReadOptions& ro,
   }
 }
 
-// TODO akanksha: Add perf_times etc.
+// TODO akanksha:
+// 1. Handle use_direct_io case which currently calls Read API.
 IOStatus RandomAccessFileReader::ReadAsync(
     FSReadRequest& req, const IOOptions& opts,
     std::function<void(const FSReadRequest&, void*)> cb, void* cb_arg,
     void** io_handle, IOHandleDeleter* del_fn,
     Env::IOPriority rate_limiter_priority) {
   if (use_direct_io()) {
+    // For direct_io, it calls Read API.
     req.status = Read(opts, req.offset, req.len, &(req.result), req.scratch,
                       nullptr /*dbg*/, rate_limiter_priority);
     cb(req, cb_arg);
     return IOStatus::OK();
   }
-  return file_->ReadAsync(req, opts, cb, cb_arg, io_handle, del_fn,
-                          nullptr /*dbg*/);
+
+  // Create a callback and populate info.
+  auto read_async_callback =
+      std::bind(&RandomAccessFileReader::ReadAsyncCallback, this,
+                std::placeholders::_1, std::placeholders::_2);
+  ReadAsyncInfo* read_async_info = new ReadAsyncInfo;
+  read_async_info->cb_ = cb;
+  read_async_info->cb_arg_ = cb_arg;
+  read_async_info->start_time_ = clock_->NowMicros();
+
+#ifndef ROCKSDB_LITE
+  if (ShouldNotifyListeners()) {
+    read_async_info->fs_start_ts_ = FileOperationInfo::StartNow();
+  }
+#endif
+
+  IOStatus s = file_->ReadAsync(req, opts, read_async_callback, read_async_info,
+                                io_handle, del_fn, nullptr /*dbg*/);
+  if (!s.ok()) {
+    delete read_async_info;
+  }
+  return s;
+}
+
+void RandomAccessFileReader::ReadAsyncCallback(const FSReadRequest& req,
+                                               void* cb_arg) {
+  ReadAsyncInfo* read_async_info = static_cast<ReadAsyncInfo*>(cb_arg);
+  assert(read_async_info);
+  assert(read_async_info->cb_);
+
+  read_async_info->cb_(req, read_async_info->cb_arg_);
+
+  // Update stats and notify listeners.
+  if (stats_ != nullptr && file_read_hist_ != nullptr) {
+    // elapsed doesn't take into account delay and overwrite as StopWatch does
+    // in Read.
+    uint64_t elapsed = clock_->NowMicros() - read_async_info->start_time_;
+    file_read_hist_->Add(elapsed);
+  }
+  if (req.status.ok()) {
+    RecordInHistogram(stats_, ASYNC_READ_BYTES, req.result.size());
+  }
+#ifndef ROCKSDB_LITE
+  if (ShouldNotifyListeners()) {
+    auto finish_ts = FileOperationInfo::FinishNow();
+    NotifyOnFileReadFinish(req.offset, req.result.size(),
+                           read_async_info->fs_start_ts_, finish_ts,
+                           req.status);
+  }
+  if (!req.status.ok()) {
+    NotifyOnIOError(req.status, FileOperationType::kRead, file_name(),
+                    req.result.size(), req.offset);
+  }
+#endif
+  RecordIOStats(stats_, file_temperature_, is_last_level_, req.result.size());
+  delete read_async_info;
 }
 }  // namespace ROCKSDB_NAMESPACE

file/random_access_file_reader.h

@@ -92,6 +92,15 @@ class RandomAccessFileReader {
   const Temperature file_temperature_;
   const bool is_last_level_;
 
+  struct ReadAsyncInfo {
+#ifndef ROCKSDB_LITE
+    FileOperationInfo::StartTimePoint fs_start_ts_;
+#endif
+    uint64_t start_time_;
+    std::function<void(const FSReadRequest&, void*)> cb_;
+    void* cb_arg_;
+  };
+
  public:
   explicit RandomAccessFileReader(
       std::unique_ptr<FSRandomAccessFile>&& raf, const std::string& _file_name,
@@ -179,5 +188,7 @@ class RandomAccessFileReader {
                      std::function<void(const FSReadRequest&, void*)> cb,
                      void* cb_arg, void** io_handle, IOHandleDeleter* del_fn,
                      Env::IOPriority rate_limiter_priority);
+
+  void ReadAsyncCallback(const FSReadRequest& req, void* cb_arg);
 };
 }  // namespace ROCKSDB_NAMESPACE

include/rocksdb/filter_policy.h

@@ -90,6 +90,19 @@ class FilterPolicy : public Customizable {
   virtual ~FilterPolicy();
   static const char* Type() { return "FilterPolicy"; }
 
+  // The name used for identifying whether a filter on disk is readable
+  // by this FilterPolicy. If this FilterPolicy is part of a family that
+  // can read each others filters, such as built-in BloomFilterPolcy and
+  // RibbonFilterPolicy, the CompatibilityName is a shared family name,
+  // while kinds of filters in the family can have distinct Customizable
+  // Names. This function is pure virtual so that wrappers around built-in
+  // policies are prompted to defer to CompatibilityName() of the wrapped
+  // policy, which is important for compatibility.
+  //
+  // For custom filter policies that are not part of a read-compatible
+  // family (rare), implementations may return Name().
+  virtual const char* CompatibilityName() const = 0;
+
   // Creates a new FilterPolicy based on the input value string and returns the
   // result The value might be an ID, and ID with properties, or an old-style
   // policy string.

include/rocksdb/statistics.h

@@ -534,6 +534,8 @@ enum Histograms : uint32_t {
   // Error handler statistics
   ERROR_HANDLER_AUTORESUME_RETRY_COUNT,
 
+  ASYNC_READ_BYTES,
+
   HISTOGRAM_ENUM_MAX,
 };
 

include/rocksdb/version.h

@@ -11,7 +11,7 @@
 
 #define ROCKSDB_MAJOR 7
 #define ROCKSDB_MINOR 1
-#define ROCKSDB_PATCH 0
+#define ROCKSDB_PATCH 1
 
 // 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

@@ -5551,7 +5551,9 @@ class HistogramTypeJni {
       case ROCKSDB_NAMESPACE::Histograms::NUM_SST_READ_PER_LEVEL:
         return 0x31;
       case ROCKSDB_NAMESPACE::Histograms::ERROR_HANDLER_AUTORESUME_RETRY_COUNT:
-        return 0x31;
+        return 0x32;
+      case ROCKSDB_NAMESPACE::Histograms::ASYNC_READ_BYTES:
+        return 0x33;
       case ROCKSDB_NAMESPACE::Histograms::HISTOGRAM_ENUM_MAX:
         // 0x1F for backwards compatibility on current minor version.
         return 0x1F;
@@ -5669,6 +5671,8 @@ class HistogramTypeJni {
       case 0x32:
         return ROCKSDB_NAMESPACE::Histograms::
             ERROR_HANDLER_AUTORESUME_RETRY_COUNT;
+      case 0x33:
+        return ROCKSDB_NAMESPACE::Histograms::ASYNC_READ_BYTES;
       case 0x1F:
         // 0x1F for backwards compatibility on current minor version.
         return ROCKSDB_NAMESPACE::Histograms::HISTOGRAM_ENUM_MAX;

java/src/main/java/org/rocksdb/HistogramType.java

@@ -180,6 +180,8 @@ public enum HistogramType {
    */
   ERROR_HANDLER_AUTORESUME_RETRY_COUNT((byte) 0x32),
 
+  ASYNC_READ_BYTES((byte) 0x33),
+
   // 0x1F for backwards compatibility on current minor version.
   HISTOGRAM_ENUM_MAX((byte) 0x1F);
 

monitoring/statistics.cc

@@ -283,6 +283,7 @@ const std::vector<std::pair<Histograms, std::string>> HistogramsNameMap = {
     {NUM_SST_READ_PER_LEVEL, "rocksdb.num.sst.read.per.level"},
     {ERROR_HANDLER_AUTORESUME_RETRY_COUNT,
      "rocksdb.error.handler.autoresume.retry.count"},
+    {ASYNC_READ_BYTES, "rocksdb.async.read.bytes"},
 };
 
 std::shared_ptr<Statistics> CreateDBStatistics() {

options/customizable_test.cc

@@ -1487,6 +1487,7 @@ class MockFilterPolicy : public FilterPolicy {
  public:
   static const char* kClassName() { return "MockFilterPolicy"; }
   const char* Name() const override { return kClassName(); }
+  const char* CompatibilityName() const override { return Name(); }
   FilterBitsBuilder* GetBuilderWithContext(
       const FilterBuildingContext&) const override {
     return nullptr;

table/block_based/block_based_table_builder.cc

@@ -1605,7 +1605,7 @@ void BlockBasedTableBuilder::WriteFilterBlock(
                 ? BlockBasedTable::kPartitionedFilterBlockPrefix
                 : BlockBasedTable::kFullFilterBlockPrefix;
     }
-    key.append(rep_->table_options.filter_policy->Name());
+    key.append(rep_->table_options.filter_policy->CompatibilityName());
     meta_index_builder->Add(key, filter_block_handle);
   }
 }

table/block_based/block_based_table_reader.cc

@@ -12,6 +12,7 @@
 #include <array>
 #include <limits>
 #include <string>
+#include <unordered_set>
 #include <utility>
 #include <vector>
 
@@ -50,6 +51,7 @@
 #include "table/block_based/block_prefix_index.h"
 #include "table/block_based/block_type.h"
 #include "table/block_based/filter_block.h"
+#include "table/block_based/filter_policy_internal.h"
 #include "table/block_based/full_filter_block.h"
 #include "table/block_based/hash_index_reader.h"
 #include "table/block_based/partitioned_filter_block.h"
@@ -897,33 +899,59 @@ Status BlockBasedTable::PrefetchIndexAndFilterBlocks(
     const BlockBasedTableOptions& table_options, const int level,
     size_t file_size, size_t max_file_size_for_l0_meta_pin,
     BlockCacheLookupContext* lookup_context) {
-  Status s;
-
   // Find filter handle and filter type
   if (rep_->filter_policy) {
-    for (auto filter_type :
-         {Rep::FilterType::kFullFilter, Rep::FilterType::kPartitionedFilter,
-          Rep::FilterType::kBlockFilter}) {
-      std::string prefix;
-      switch (filter_type) {
-        case Rep::FilterType::kFullFilter:
-          prefix = kFullFilterBlockPrefix;
+    auto name = rep_->filter_policy->CompatibilityName();
+    bool builtin_compatible =
+        strcmp(name, BuiltinFilterPolicy::kCompatibilityName()) == 0;
+
+    for (const auto& [filter_type, prefix] :
+         {std::make_pair(Rep::FilterType::kFullFilter, kFullFilterBlockPrefix),
+          std::make_pair(Rep::FilterType::kPartitionedFilter,
+                         kPartitionedFilterBlockPrefix),
+          std::make_pair(Rep::FilterType::kBlockFilter, kFilterBlockPrefix)}) {
+      if (builtin_compatible) {
+        // This code is only here to deal with a hiccup in early 7.0.x where
+        // there was an unintentional name change in the SST files metadata.
+        // It should be OK to remove this in the future (late 2022) and just
+        // have the 'else' code.
+        // NOTE: the test:: names below are likely not needed but included
+        // out of caution
+        static const std::unordered_set<std::string> kBuiltinNameAndAliases = {
+            BuiltinFilterPolicy::kCompatibilityName(),
+            test::LegacyBloomFilterPolicy::kClassName(),
+            test::FastLocalBloomFilterPolicy::kClassName(),
+            test::Standard128RibbonFilterPolicy::kClassName(),
+            DeprecatedBlockBasedBloomFilterPolicy::kClassName(),
+            BloomFilterPolicy::kClassName(),
+            RibbonFilterPolicy::kClassName(),
+        };
+
+        // For efficiency, do a prefix seek and see if the first match is
+        // good.
+        meta_iter->Seek(prefix);
+        if (meta_iter->status().ok() && meta_iter->Valid()) {
+          Slice key = meta_iter->key();
+          if (key.starts_with(prefix)) {
+            key.remove_prefix(prefix.size());
+            if (kBuiltinNameAndAliases.find(key.ToString()) !=
+                kBuiltinNameAndAliases.end()) {
+              Slice v = meta_iter->value();
+              Status s = rep_->filter_handle.DecodeFrom(&v);
+              if (s.ok()) {
+                rep_->filter_type = filter_type;
+                break;
+              }
+            }
+          }
+        }
+      } else {
+        std::string filter_block_key = prefix + name;
+        if (FindMetaBlock(meta_iter, filter_block_key, &rep_->filter_handle)
+                .ok()) {
+          rep_->filter_type = filter_type;
           break;
-        case Rep::FilterType::kPartitionedFilter:
-          prefix = kPartitionedFilterBlockPrefix;
-          break;
-        case Rep::FilterType::kBlockFilter:
-          prefix = kFilterBlockPrefix;
-          break;
-        default:
-          assert(0);
-      }
-      std::string filter_block_key = prefix;
-      filter_block_key.append(rep_->filter_policy->Name());
-      if (FindMetaBlock(meta_iter, filter_block_key, &rep_->filter_handle)
-              .ok()) {
-        rep_->filter_type = filter_type;
-        break;
+        }
       }
     }
   }
@@ -932,8 +960,8 @@ Status BlockBasedTable::PrefetchIndexAndFilterBlocks(
          rep_->index_type == BlockBasedTableOptions::kTwoLevelIndexSearch);
 
   // Find compression dictionary handle
-  s = FindOptionalMetaBlock(meta_iter, kCompressionDictBlockName,
-                            &rep_->compression_dict_handle);
+  Status s = FindOptionalMetaBlock(meta_iter, kCompressionDictBlockName,
+                                   &rep_->compression_dict_handle);
   if (!s.ok()) {
     return s;
   }

table/block_based/block_based_table_reader.h

@@ -654,10 +654,10 @@ struct BlockBasedTable::Rep {
                                 std::unique_ptr<FilePrefetchBuffer>* fpb,
                                 bool implicit_auto_readahead,
                                 bool async_io) const {
-    fpb->reset(new FilePrefetchBuffer(readahead_size, max_readahead_size,
-                                      !ioptions.allow_mmap_reads /* enable */,
-                                      false /* track_min_offset */,
-                                      implicit_auto_readahead, async_io));
+    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()));
   }
 
   void CreateFilePrefetchBufferIfNotExists(

table/block_based/filter_policy.cc

@@ -1325,6 +1325,16 @@ bool BuiltinFilterPolicy::IsInstanceOf(const std::string& name) const {
   }
 }
 
+static const char* kBuiltinFilterMetadataName = "rocksdb.BuiltinBloomFilter";
+
+const char* BuiltinFilterPolicy::kCompatibilityName() {
+  return kBuiltinFilterMetadataName;
+}
+
+const char* BuiltinFilterPolicy::CompatibilityName() const {
+  return kBuiltinFilterMetadataName;
+}
+
 BloomLikeFilterPolicy::BloomLikeFilterPolicy(double bits_per_key)
     : warned_(false), aggregate_rounding_balance_(0) {
   // Sanitize bits_per_key
@@ -1372,7 +1382,7 @@ bool BloomLikeFilterPolicy::IsInstanceOf(const std::string& name) const {
 }
 
 const char* ReadOnlyBuiltinFilterPolicy::kClassName() {
-  return "rocksdb.BuiltinBloomFilter";
+  return kBuiltinFilterMetadataName;
 }
 
 const char* DeprecatedBlockBasedBloomFilterPolicy::kClassName() {

table/block_based/filter_policy_internal.h

@@ -135,6 +135,9 @@ class BuiltinFilterPolicy : public FilterPolicy {
   FilterBitsReader* GetFilterBitsReader(const Slice& contents) const override;
   static const char* kClassName();
   bool IsInstanceOf(const std::string& id) const override;
+  // All variants of BuiltinFilterPolicy can read each others filters.
+  const char* CompatibilityName() const override;
+  static const char* kCompatibilityName();
 
  public:  // new
   // An internal function for the implementation of

table/block_based/full_filter_block_test.cc

@@ -84,6 +84,7 @@ class TestFilterBitsReader : public FilterBitsReader {
 class TestHashFilter : public FilterPolicy {
  public:
   const char* Name() const override { return "TestHashFilter"; }
+  const char* CompatibilityName() const override { return Name(); }
 
   FilterBitsBuilder* GetBuilderWithContext(
       const FilterBuildingContext&) const override {

table/block_fetcher.cc

@@ -74,8 +74,7 @@ inline bool BlockFetcher::TryGetFromPrefetchBuffer() {
       if (read_options_.async_io) {
         read_from_prefetch_buffer = prefetch_buffer_->TryReadFromCacheAsync(
             opts, file_, handle_.offset(), block_size_with_trailer_, &slice_,
-            &io_s, read_options_.rate_limiter_priority, for_compaction_,
-            ioptions_.fs.get());
+            &io_s, read_options_.rate_limiter_priority, for_compaction_);
       } else {
         read_from_prefetch_buffer = prefetch_buffer_->TryReadFromCache(
             opts, file_, handle_.offset(), block_size_with_trailer_, &slice_,