Update version and HISTORY.md for 11.4.3 release

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

Read-only DB instances can share a directory with a live writer, so their view of live files is only a snapshot from the time they opened. After read-only DB open began setting `opened_successfully_` for its intended open-success semantics, that state had an unintended side effect: the common close path treated a successful read-only open like a read-write open and ran obsolete-file cleanup. If the writer created or made files live after the read-only handle opened, that cleanup could use the stale read-only live set and delete files still needed by the writer.

This change records whether a `DBImpl` was opened read-only and skips close-time obsolete-file cleanup for read-only DBs instead of simply keeping `opened_successfully_=false`, which could be confusing and easily mistaken in the future again.

For completeness I also updated other callsites of `opened_successfully_`, but those should not be real bugs as ReadOnly DBs do not run flushes/compactions or write to WALs.

Reviewed By: xingbowang

Differential Revision: D109622445

fbshipit-source-id: d7be4b20fce86ccb218a63ac6f5b707b316aac79

HISTORY.md

@@ -1,6 +1,41 @@
 # Rocksdb Change Log
 > NOTE: Entries for next release do not go here. Follow instructions in `unreleased_history/README.txt`
 
+## 11.4.3 (06/24/2026)
+### Bug Fixes
+* Fixed a bug where closing a read-only DB instance could delete live SST files created by a concurrent read-write DB sharing the same directory.
+
+## 11.4.2 (06/08/2026)
+### Public API Changes
+* Added experimental read-scoped block buffer provider API, configured through `ReadOptions::read_scoped_block_buffer_provider`, for supported block-based table iterator scans and MultiScan reads to use caller-provided read-scoped storage for final data-block contents. When configured, supported provider-backed scan data-block reads bypass the data-block cache. The provider is ignored when mmap reads are enabled. RocksDB may still use ordinary temporary scratch for serialized block bytes, such as when a block may be compressed. Get and MultiGet do not currently provide API guarantees for this provider.
+
+### Bug Fixes
+* Fixed a use after free bug in Multiscan when async IO is used
+
+## 11.4.1 (06/05/2026)
+* No-op patch used to sync with internal fb version.
+
+## 11.4.0 (06/02/2026)
+### Public API Changes
+* Added `rocksdb_options_set_memtable_batch_lookup_optimization()` and `rocksdb_options_get_memtable_batch_lookup_optimization()` to the C API, exposing the existing `AdvancedColumnFamilyOptions::memtable_batch_lookup_optimization` field. This allows C API users (and downstream language bindings) to enable the skip-list memtable's batch-lookup optimization for `MultiGet`, which caches the search path between consecutive keys and reduces per-key cost from O(log N) to O(log d) where d is the distance between consecutive keys.
+* Added `rocksdb_readoptions_set_optimize_multiget_for_io()` and `rocksdb_readoptions_get_optimize_multiget_for_io()` to the C API, exposing the existing `ReadOptions::optimize_multiget_for_io` field. This allows C API users (and downstream language bindings) to opt out of the multi-level parallel MultiGet path, which only takes effect when the library is built with `USE_COROUTINES`.
+* Added `rocksdb_block_based_table_index_block_search_type_auto` enum constant and the `rocksdb_block_based_options_set_uniform_cv_threshold()` setter to the C API. The constant exposes `BlockSearchType::kAuto`, and the setter exposes `BlockBasedTableOptions::uniform_cv_threshold`. Both are required for `kAuto` index-block search to take effect from the C API: without setting `uniform_cv_threshold >= 0`, the per-block `is_uniform` footer bit is never written, and `kAuto` falls back to binary search at read time.
+* Added `EventListener::OnDBShutdownBegin`, a callback that fires once when a DB begins shutdown, including when RocksDB cleans up a failed `DB::Open()` attempt.
+* Added a new PerfContext counter `blob_cache_read_byte` for blob cache bytes read.
+
+### Behavior Changes
+* Remote compaction now falls back to local compaction when the primary cannot deserialize a successful `CompactionServiceResult` before installing any remote output files.
+* StringToMap (rocksdb/convenience.h) now preserves the outer braces of nested values so each map entry is in self-contained form and can be embedded directly into another `key=value;` string (e.g. SetOptions) without losing inner ';' delimiters. A new symmetric MapToString utility is provided.
+
+### Bug Fixes
+* Fixed a bug where `AbortAllCompactions()` could leave automatic compaction work unscheduled when aborting before the background worker picked it, causing compaction and write stalls until DB restart.
+* Fix a bug where db had a false positive compaction corruption error due to remote compaction service result with `has_accurate_num_input_records=false` not being serialized.
+* Fixed WritePrepared TransactionDB cleanup after retryable commit or rollback write failures so prepared transactions remain rollbackable instead of leaving unresolved prepared state.
+* Fix a rare corruption bug for tiered compaction that incorrectly moved last level range tombstones into proximal level. This corruption error is surfaced when force_consistency_checks is enabled.
+
+### Performance Improvements
+* Reduce the likelihood of user-facing metadata operations blocking on MANIFEST rotation when file creation is slow, such as on remote storage. MANIFEST write batches containing foreground edits from `CreateColumnFamily()`, `DropColumnFamily()`, `CreateColumnFamilyWithImport()`, `IngestExternalFile()`, and `DeleteFilesInRanges()` now get a relaxed file size threshold for triggering MANIFEST rotation; background-only MANIFEST write batches, such as compaction and flush, continue to use the normal threshold. This change should make auto-tuning manifest file size more attractive (see `max_manifest_file_size` and `max_manifest_space_amp_pct` options).
+
 ## 11.3.0 (05/15/2026)
 ### New Features
 * Add experimental DB option `async_wal_precreate` to precreate the next WAL file in a background thread and reduce foreground WAL rotation latency. The option is sanitized to false when WAL recycling is enabled.

db/blob/blob_file_reader.cc

@@ -20,6 +20,7 @@
 #include "table/format.h"
 #include "table/multiget_context.h"
 #include "test_util/sync_point.h"
+#include "util/aligned_buffer.h"
 #include "util/compression.h"
 #include "util/stop_watch.h"
 
@@ -165,7 +166,7 @@ Status BlobFileReader::ReadHeader(const RandomAccessFileReader* file_reader,
 
   Slice header_slice;
   Buffer buf;
-  AlignedBuf aligned_buf;
+  AlignedBuffer direct_io_buffer;
 
   {
     TEST_SYNC_POINT("BlobFileReader::ReadHeader:ReadFromFile");
@@ -175,7 +176,7 @@ Status BlobFileReader::ReadHeader(const RandomAccessFileReader* file_reader,
 
     const Status s =
         ReadFromFile(file_reader, read_options, read_offset, read_size,
-                     statistics, &header_slice, &buf, &aligned_buf);
+                     statistics, &header_slice, &buf, &direct_io_buffer);
     if (!s.ok()) {
       return s;
     }
@@ -216,7 +217,7 @@ Status BlobFileReader::ReadFooter(const RandomAccessFileReader* file_reader,
 
   Slice footer_slice;
   Buffer buf;
-  AlignedBuf aligned_buf;
+  AlignedBuffer direct_io_buffer;
 
   {
     TEST_SYNC_POINT("BlobFileReader::ReadFooter:ReadFromFile");
@@ -226,7 +227,7 @@ Status BlobFileReader::ReadFooter(const RandomAccessFileReader* file_reader,
 
     const Status s =
         ReadFromFile(file_reader, read_options, read_offset, read_size,
-                     statistics, &footer_slice, &buf, &aligned_buf);
+                     statistics, &footer_slice, &buf, &direct_io_buffer);
     if (!s.ok()) {
       return s;
     }
@@ -257,10 +258,11 @@ Status BlobFileReader::ReadFromFile(const RandomAccessFileReader* file_reader,
                                     const ReadOptions& read_options,
                                     uint64_t read_offset, size_t read_size,
                                     Statistics* statistics, Slice* slice,
-                                    Buffer* buf, AlignedBuf* aligned_buf) {
+                                    Buffer* buf,
+                                    AlignedBuffer* direct_io_buffer) {
   assert(slice);
   assert(buf);
-  assert(aligned_buf);
+  assert(direct_io_buffer);
 
   assert(file_reader);
 
@@ -277,15 +279,15 @@ Status BlobFileReader::ReadFromFile(const RandomAccessFileReader* file_reader,
 
   if (file_reader->use_direct_io()) {
     constexpr char* scratch = nullptr;
+    AlignedBufferAllocationContext direct_io_context{direct_io_buffer};
 
     s = file_reader->Read(io_options, read_offset, read_size, slice, scratch,
-                          aligned_buf, &dbg);
+                          &direct_io_context, &dbg);
   } else {
     buf->reset(new char[read_size]);
-    constexpr AlignedBuf* aligned_scratch = nullptr;
 
     s = file_reader->Read(io_options, read_offset, read_size, slice, buf->get(),
-                          aligned_scratch, &dbg);
+                          nullptr, &dbg);
   }
 
   if (!s.ok()) {
@@ -349,7 +351,7 @@ Status BlobFileReader::GetBlob(
 
   Slice record_slice;
   Buffer buf;
-  AlignedBuf aligned_buf;
+  AlignedBuffer direct_io_buffer;
 
   bool prefetched = false;
 
@@ -379,7 +381,7 @@ Status BlobFileReader::GetBlob(
     const Status s =
         ReadFromFile(file_reader_.get(), read_options, record_offset,
                      static_cast<size_t>(record_size), statistics_,
-                     &record_slice, &buf, &aligned_buf);
+                     &record_slice, &buf, &direct_io_buffer);
     if (!s.ok()) {
       return s;
     }
@@ -477,7 +479,7 @@ void BlobFileReader::MultiGetBlob(
   }
 
   Buffer buf;
-  AlignedBuf aligned_buf;
+  AlignedBuffer direct_io_buffer;
 
   Status s;
   bool direct_io = file_reader_->use_direct_io();
@@ -500,8 +502,9 @@ void BlobFileReader::MultiGetBlob(
   IODebugContext dbg;
   s = file_reader_->PrepareIOOptions(read_options, opts, &dbg);
   if (s.ok()) {
+    AlignedBufferAllocationContext direct_io_context{&direct_io_buffer};
     s = file_reader_->MultiRead(opts, read_reqs.data(), read_reqs.size(),
-                                direct_io ? &aligned_buf : nullptr, &dbg);
+                                &direct_io_context, &dbg);
   }
   if (!s.ok()) {
     for (auto& req : read_reqs) {

db/blob/blob_file_reader.h

@@ -108,7 +108,7 @@ class BlobFileReader {
                              const ReadOptions& read_options,
                              uint64_t read_offset, size_t read_size,
                              Statistics* statistics, Slice* slice, Buffer* buf,
-                             AlignedBuf* aligned_buf);
+                             AlignedBuffer* direct_io_buffer);
 
   static Status VerifyBlob(const Slice& record_slice, const Slice& user_key,
                            uint64_t value_size);

db/db_impl/db_impl.cc

@@ -175,6 +175,7 @@ DBImpl::DBImpl(const DBOptions& options, const std::string& dbname,
                const bool seq_per_batch, const bool batch_per_txn,
                bool read_only)
     : dbname_(dbname),
+      read_only_(read_only),
       own_info_log_(options.info_log == nullptr),
       initial_db_options_(SanitizeOptions(dbname, options, read_only,
                                           &init_logger_creation_s_)),
@@ -671,7 +672,8 @@ void DBImpl::UnregisterBlobDirectWriteColumnFamily() {
 Status DBImpl::MaybeWriteWalMarkersToManifestOnClose() {
   mutex_.AssertHeld();
   if (!mutable_db_options_.optimize_manifest_for_recovery ||
-      !opened_successfully_ || versions_ == nullptr || logs_.empty()) {
+      !opened_successfully_ || read_only_ || versions_ == nullptr ||
+      logs_.empty()) {
     return Status::OK();
   }
 
@@ -873,7 +875,7 @@ Status DBImpl::CloseHelper() {
   // manifest file), it is not able to identify live files correctly. As a
   // result, all "live" files can get deleted by accident. However, corrupted
   // manifest is recoverable by RepairDB().
-  if (opened_successfully_) {
+  if (opened_successfully_ && !read_only_) {
     JobContext job_context(next_job_id_.fetch_add(1));
     FindObsoleteFiles(&job_context, true);
 

db/db_impl/db_impl.h

@@ -1374,6 +1374,7 @@ class DBImpl : public DB {
 
  protected:
   const std::string dbname_;
+  const bool read_only_;
   // TODO(peterd): unify with VersionSet::db_id_
   std::string db_id_;
   // db_session_id_ is an identifier that gets reset

db/db_impl/db_impl_compaction_flush.cc

@@ -3223,7 +3223,7 @@ void DBImpl::ResumeAllCompactions() {
 void DBImpl::MaybeScheduleFlushOrCompaction() {
   mutex_.AssertHeld();
   TEST_SYNC_POINT("DBImpl::MaybeScheduleFlushOrCompaction:Start");
-  if (!opened_successfully_) {
+  if (!opened_successfully_ || read_only_) {
     // Compaction may introduce data race to DB open
     return;
   }

db/db_test.cc

@@ -8071,6 +8071,41 @@ INSTANTIATE_TEST_CASE_P(OpenFilesAsync, OpenFilesAsyncTest,
                                            ::testing::Values(-1, 10),
                                            ::testing::Bool()));
 
+TEST_F(DBTest, ReadOnlyCloseDoesNotDeleteWriterFiles) {
+  Options options = CurrentOptions();
+  options.create_if_missing = true;
+  options.disable_auto_compactions = true;
+  DestroyAndReopen(options);
+
+  ASSERT_OK(Put("before", "value"));
+  ASSERT_OK(Flush());
+
+  std::unique_ptr<DB> read_only_db;
+  ASSERT_OK(DB::OpenForReadOnly(options, dbname_, &read_only_db));
+
+  ASSERT_OK(Put("after", "value"));
+  ASSERT_OK(Flush());
+
+  std::vector<LiveFileMetaData> live_files;
+  db_->GetLiveFilesMetaData(&live_files);
+  ASSERT_GE(live_files.size(), 2);
+
+  std::string newest_sst_path;
+  uint64_t newest_file_number = 0;
+  for (const auto& live_file : live_files) {
+    if (live_file.file_number > newest_file_number) {
+      newest_file_number = live_file.file_number;
+      newest_sst_path = live_file.directory + "/" + live_file.relative_filename;
+    }
+  }
+  ASSERT_FALSE(newest_sst_path.empty());
+  ASSERT_OK(env_->FileExists(newest_sst_path));
+
+  read_only_db.reset();
+
+  ASSERT_OK(env_->FileExists(newest_sst_path));
+}
+
 // Test mix of races with async file open, reads, compactions
 TEST_P(OpenFilesAsyncTest, ConcurrentFileAccess) {
   Options options = CurrentOptions();

db_stress_tool/db_stress_common.h

@@ -466,6 +466,7 @@ DECLARE_uint32(ingest_wbwi_one_in);
 DECLARE_bool(universal_reduce_file_locking);
 DECLARE_bool(use_multiscan);
 DECLARE_bool(multiscan_use_async_io);
+DECLARE_bool(read_scoped_block_buffer_provider);
 DECLARE_uint64(multiscan_max_prefetch_memory_bytes);
 
 // Compaction deletion trigger declarations for stress testing

db_stress_tool/db_stress_gflags.cc

@@ -1705,6 +1705,10 @@ DEFINE_bool(use_multiscan, false,
 DEFINE_bool(multiscan_use_async_io, false,
             "If set, enable async_io for MultiScan operations.");
 
+DEFINE_bool(read_scoped_block_buffer_provider, false,
+            "If set, configure ReadOptions::read_scoped_block_buffer_provider "
+            "with a stress-test provider for supported scan reads.");
+
 DEFINE_uint64(multiscan_max_prefetch_memory_bytes, 0,
               "If non-zero, sets the max_prefetch_memory_bytes on the "
               "IODispatcher used for MultiScan. This limits the total memory "

db_stress_tool/db_stress_test_base.cc

@@ -9,7 +9,10 @@
 //
 
 #include <ios>
+#include <memory>
+#include <mutex>
 #include <thread>
+#include <unordered_set>
 
 #include "db_stress_tool/db_stress_compression_manager.h"
 #include "db_stress_tool/db_stress_listener.h"
@@ -32,11 +35,13 @@
 #include "rocksdb/io_dispatcher.h"
 #include "rocksdb/secondary_cache.h"
 #include "rocksdb/sst_file_manager.h"
+#include "rocksdb/table.h"
 #include "rocksdb/table_properties.h"
 #include "rocksdb/types.h"
 #include "rocksdb/utilities/object_registry.h"
 #include "rocksdb/utilities/write_batch_with_index.h"
 #include "test_util/testutil.h"
+#include "util/aligned_buffer.h"
 #include "util/cast_util.h"
 #include "util/simple_mixed_compressor.h"
 #include "utilities/backup/backup_engine_impl.h"
@@ -48,6 +53,120 @@ namespace ROCKSDB_NAMESPACE {
 
 namespace {
 
+class StressReadScopedBlockBufferProvider
+    : public ReadScopedBlockBufferProvider {
+ public:
+  StressReadScopedBlockBufferProvider() : state_(std::make_shared<State>()) {}
+  StressReadScopedBlockBufferProvider(
+      const StressReadScopedBlockBufferProvider&) = delete;
+  StressReadScopedBlockBufferProvider& operator=(
+      const StressReadScopedBlockBufferProvider&) = delete;
+  StressReadScopedBlockBufferProvider(StressReadScopedBlockBufferProvider&&) =
+      delete;
+  StressReadScopedBlockBufferProvider& operator=(
+      StressReadScopedBlockBufferProvider&&) = delete;
+
+  ~StressReadScopedBlockBufferProvider() override {
+    state_->ValidateNoLiveAllocations();
+  }
+
+  Status Allocate(size_t size, size_t alignment, Lease* out) override {
+    Check(out != nullptr, "lease output is nullptr");
+    Check(size > 0, "allocation size is zero");
+    Check(alignment > 0 && (alignment & (alignment - 1)) == 0,
+          "alignment must be a power of two");
+
+    auto* allocation = new Allocation();
+    allocation->state = state_;
+    allocation->requested_size = size;
+    allocation->requested_alignment = alignment;
+    allocation->storage.Alignment(alignment);
+    allocation->storage.AllocateNewBuffer(size);
+    allocation->data = allocation->storage.BufferStart();
+    allocation->size = allocation->storage.Capacity();
+
+    Check(allocation->data != nullptr, "provider returned null data");
+    Check(allocation->size >= size, "provider returned short allocation");
+    Check(AlignedBuffer::isAligned(allocation->data, alignment),
+          "provider returned misaligned data");
+    Check(
+        alignment == 1 || AlignedBuffer::isAligned(allocation->size, alignment),
+        "provider returned misaligned direct-I/O size");
+
+    {
+      std::lock_guard<std::mutex> lock(state_->mutex);
+      Check(state_->live_allocations.insert(allocation).second,
+            "duplicate allocation registration");
+      state_->live_bytes += allocation->size;
+      ++state_->total_allocations;
+    }
+
+    out->data = allocation->data;
+    out->size = allocation->size;
+    out->cleanup.Allocate();
+    out->cleanup->RegisterCleanup(&ReleaseAllocation, allocation, nullptr);
+    return Status::OK();
+  }
+
+ private:
+  struct Allocation;
+
+  struct State {
+    std::mutex mutex;
+    std::unordered_set<Allocation*> live_allocations;
+    size_t live_bytes = 0;
+    uint64_t total_allocations = 0;
+
+    void ValidateNoLiveAllocations() {
+      std::lock_guard<std::mutex> lock(mutex);
+      Check(live_allocations.empty(),
+            "provider destroyed with live allocations");
+      Check(live_bytes == 0, "provider destroyed with live bytes");
+    }
+  };
+
+  struct Allocation {
+    std::shared_ptr<State> state;
+    size_t requested_size = 0;
+    size_t requested_alignment = 1;
+    char* data = nullptr;
+    size_t size = 0;
+    AlignedBuffer storage;
+  };
+
+  static void Check(bool condition, const char* message) {
+    if (!condition) {
+      fprintf(stderr, "ReadScopedBlockBufferProvider invariant failed: %s\n",
+              message);
+      std::abort();
+    }
+  }
+
+  static void ReleaseAllocation(void* arg1, void* /*arg2*/) {
+    auto* allocation = static_cast<Allocation*>(arg1);
+    Check(allocation != nullptr, "cleanup received null allocation");
+    Check(allocation->state != nullptr, "cleanup received null state");
+    Check(allocation->data != nullptr, "cleanup received null data");
+    Check(allocation->size >= allocation->requested_size,
+          "cleanup received short allocation");
+    Check(AlignedBuffer::isAligned(allocation->data,
+                                   allocation->requested_alignment),
+          "cleanup received misaligned data");
+
+    {
+      std::lock_guard<std::mutex> lock(allocation->state->mutex);
+      Check(allocation->state->live_allocations.erase(allocation) == 1,
+            "cleanup for unknown or duplicate allocation");
+      Check(allocation->state->live_bytes >= allocation->size,
+            "cleanup underflowed live bytes");
+      allocation->state->live_bytes -= allocation->size;
+    }
+    delete allocation;
+  }
+
+  std::shared_ptr<State> state_;
+};
+
 std::shared_ptr<const FilterPolicy> CreateFilterPolicy() {
   if (FLAGS_bloom_bits < 0) {
     return BlockBasedTableOptions().filter_policy;
@@ -1207,6 +1326,14 @@ void StressTest::OperateDb(ThreadState* thread) {
   if (FLAGS_use_trie_index && !FLAGS_use_udi_as_primary_index && udi_factory_) {
     read_opts.table_index_factory = udi_factory_.get();
   }
+  std::unique_ptr<StressReadScopedBlockBufferProvider>
+      read_scoped_block_buffer_provider;
+  if (FLAGS_read_scoped_block_buffer_provider) {
+    read_scoped_block_buffer_provider =
+        std::make_unique<StressReadScopedBlockBufferProvider>();
+    read_opts.read_scoped_block_buffer_provider =
+        read_scoped_block_buffer_provider.get();
+  }
   WriteOptions write_opts;
   if (FLAGS_rate_limit_auto_wal_flush) {
     write_opts.rate_limiter_priority = Env::IO_USER;

env/env_test.cc

@@ -295,6 +295,46 @@ class EnvPosixTestWithParam
     }
   }
 
+  // ReserveThreads() returns the number of threads observed waiting at that
+  // instant. When this test runs in parallel with many other CPU-heavy tests,
+  // worker-thread scheduling can vary enough that the sync points prove key
+  // transitions were reached before waiting-thread accounting has fully
+  // settled. Use this helper only for positive exact-reservation checks; it
+  // releases any partial reservation before retrying so the retry does not
+  // perturb test state.
+  testing::AssertionResult ReserveThreadsEventually(int expected, int requested,
+                                                    Env::Priority priority,
+                                                    int wait_micros) {
+    constexpr int kRetryMicros = 1000;
+    int last_reserved = -1;
+    for (int waited_micros = 0; waited_micros <= wait_micros;
+         waited_micros += kRetryMicros) {
+      int reserved = env_->ReserveThreads(requested, priority);
+      if (reserved == expected) {
+        return testing::AssertionSuccess();
+      }
+      if (reserved > 0) {
+        int released = env_->ReleaseThreads(reserved, priority);
+        if (released != reserved) {
+          return testing::AssertionFailure()
+                 << "ReserveThreads(" << requested << ") returned " << reserved
+                 << ", but ReleaseThreads(" << reserved << ") released "
+                 << released;
+        }
+      }
+      if (reserved > expected) {
+        return testing::AssertionFailure()
+               << "ReserveThreads(" << requested << ") returned " << reserved
+               << ", more than expected " << expected;
+      }
+      last_reserved = reserved;
+      Env::Default()->SleepForMicroseconds(kRetryMicros);
+    }
+    return testing::AssertionFailure()
+           << "ReserveThreads(" << requested << ") returned " << last_reserved
+           << " after waiting " << wait_micros << "us, expected " << expected;
+  }
+
   ~EnvPosixTestWithParam() override { WaitThreadPoolsEmpty(); }
 };
 
@@ -1022,7 +1062,7 @@ TEST_P(EnvPosixTestWithParam, ReserveThreads) {
   TEST_SYNC_POINT("EnvTest::ReserveThreads:2");
   TEST_SYNC_POINT("EnvTest::ReserveThreads:3");
   // Reserve 2 threads
-  ASSERT_EQ(2, env_->ReserveThreads(2, Env::Priority::HIGH));
+  ASSERT_TRUE(ReserveThreadsEventually(2, 2, Env::Priority::HIGH, kWaitMicros));
 
   // Schedule 3 tasks. Task 0 running (in this context, doing
   // SleepingBackgroundTask); Task 1, 2 waiting; 3 reserved threads.
@@ -1051,7 +1091,7 @@ TEST_P(EnvPosixTestWithParam, ReserveThreads) {
   // Add sync point to ensure the 4th thread starts
   TEST_SYNC_POINT("EnvTest::ReserveThreads:4");
   // As the thread pool is expanded, we can reserve one more thread
-  ASSERT_EQ(1, env_->ReserveThreads(3, Env::Priority::HIGH));
+  ASSERT_TRUE(ReserveThreadsEventually(1, 3, Env::Priority::HIGH, kWaitMicros));
   // No more threads can be reserved
   ASSERT_EQ(0, env_->ReserveThreads(3, Env::Priority::HIGH));
 
@@ -1070,7 +1110,7 @@ TEST_P(EnvPosixTestWithParam, ReserveThreads) {
   // Add sync point to ensure the number of waiting threads increases
   TEST_SYNC_POINT("EnvTest::ReserveThreads:5");
   // 1 more thread can be reserved
-  ASSERT_EQ(1, env_->ReserveThreads(3, Env::Priority::HIGH));
+  ASSERT_TRUE(ReserveThreadsEventually(1, 3, Env::Priority::HIGH, kWaitMicros));
   // 2 reserved threads now
 
   // Currently, two threads are blocked since the number of waiting

file/file_prefetch_buffer.cc

@@ -103,7 +103,7 @@ Status FilePrefetchBuffer::Read(BufferInfo* buf, const IOOptions& opts,
   } else {
     to_buf = buf->buffer_.BufferStart() + aligned_useful_len;
     s = reader->Read(opts, start_offset + aligned_useful_len, read_len, &result,
-                     to_buf, /*aligned_buf=*/nullptr);
+                     to_buf);
   }
 
 #ifndef NDEBUG
@@ -165,7 +165,7 @@ Status FilePrefetchBuffer::ReadAsync(BufferInfo* buf, const IOOptions& opts,
     // Fall back to synchronous read so the buffer is populated inline
     // and callers proceed transparently.
     s = reader->Read(opts, start_offset, read_len, &result,
-                     buf->buffer_.BufferStart(), /*aligned_buf=*/nullptr);
+                     buf->buffer_.BufferStart());
     if (s.ok()) {
       buf->buffer_.Size(buf->CurrentSize() + result.size());
       if (usage_ == FilePrefetchBufferUsage::kUserScanPrefetch) {

file/file_prefetch_buffer.h

@@ -527,7 +527,9 @@ class FilePrefetchBuffer {
     read_req.offset = offset;
     read_req.len = n;
     read_req.scratch = nullptr;
-    IOStatus s = reader->MultiRead(opts, &read_req, 1, nullptr);
+    AlignedBuffer direct_io_buffer;
+    AlignedBufferAllocationContext direct_io_context{&direct_io_buffer};
+    IOStatus s = reader->MultiRead(opts, &read_req, 1, &direct_io_context);
     if (!s.ok()) {
       return s;
     }

file/random_access_file_reader.cc

@@ -16,6 +16,7 @@
 #include "monitoring/histogram.h"
 #include "monitoring/iostats_context_imp.h"
 #include "port/port.h"
+#include "rocksdb/io_status.h"
 #include "table/format.h"
 #include "test_util/sync_point.h"
 #include "util/random.h"
@@ -128,11 +129,17 @@ IOStatus RandomAccessFileReader::Create(
   return io_s;
 }
 
-IOStatus RandomAccessFileReader::Read(const IOOptions& opts, uint64_t offset,
-                                      size_t n, Slice* result, char* scratch,
-                                      AlignedBuf* aligned_buf,
-                                      IODebugContext* dbg) const {
-  (void)aligned_buf;
+IOStatus RandomAccessFileReader::Read(
+    const IOOptions& opts, uint64_t offset, size_t n, Slice* result,
+    char* scratch, AlignedBufferAllocationContext* direct_io_buffer_context,
+    IODebugContext* dbg) const {
+  AlignedBuffer* direct_io_buffer = direct_io_buffer_context != nullptr
+                                        ? direct_io_buffer_context->buffer
+                                        : nullptr;
+  const AlignedBuffer::Allocator* direct_io_allocator =
+      direct_io_buffer_context != nullptr ? direct_io_buffer_context->allocator
+                                          : nullptr;
+  assert(direct_io_buffer_context == nullptr || direct_io_buffer != nullptr);
   const Env::IOPriority rate_limiter_priority = opts.rate_limiter_priority;
 
   TEST_SYNC_POINT_CALLBACK("RandomAccessFileReader::Read", nullptr);
@@ -151,7 +158,7 @@ IOStatus RandomAccessFileReader::Read(const IOOptions& opts, uint64_t offset,
   uint64_t elapsed = 0;
   size_t alignment = file_->GetRequiredBufferAlignment();
   bool is_aligned = false;
-  if (scratch != nullptr) {
+  if (direct_io_buffer == nullptr && scratch != nullptr) {
     // Check if offset, length and buffer are aligned.
     is_aligned = (offset & (alignment - 1)) == 0 &&
                  (n & (alignment - 1)) == 0 &&
@@ -172,17 +179,25 @@ IOStatus RandomAccessFileReader::Read(const IOOptions& opts, uint64_t offset,
       size_t read_size =
           Roundup(static_cast<size_t>(offset + n), alignment) - aligned_offset;
       AlignedBuffer buf;
-      buf.Alignment(alignment);
-      buf.AllocateNewBuffer(read_size);
-      while (buf.CurrentSize() < read_size) {
+      AlignedBuffer* aligned_read_buffer =
+          direct_io_buffer != nullptr ? direct_io_buffer : &buf;
+      aligned_read_buffer->Alignment(alignment);
+      Status allocate_status = aligned_read_buffer->AllocateNewBuffer(
+          read_size, direct_io_allocator);
+      if (!allocate_status.ok()) {
+        io_s = status_to_io_status(std::move(allocate_status));
+      }
+      char* aligned_scratch = aligned_read_buffer->BufferStart();
+      size_t current_size = 0;
+      while (io_s.ok() && current_size < read_size) {
         size_t allowed;
         if (rate_limiter_priority != Env::IO_TOTAL &&
             rate_limiter_ != nullptr) {
           allowed = rate_limiter_->RequestToken(
-              buf.Capacity() - buf.CurrentSize(), buf.Alignment(),
-              rate_limiter_priority, stats_, RateLimiter::OpType::kRead);
+              read_size - current_size, alignment, rate_limiter_priority,
+              stats_, RateLimiter::OpType::kRead);
         } else {
-          assert(buf.CurrentSize() == 0);
+          assert(current_size == 0);
           allowed = read_size;
         }
         Slice tmp;
@@ -191,7 +206,7 @@ IOStatus RandomAccessFileReader::Read(const IOOptions& opts, uint64_t offset,
         uint64_t orig_offset = 0;
         if (ShouldNotifyListeners()) {
           start_ts = FileOperationInfo::StartNow();
-          orig_offset = aligned_offset + buf.CurrentSize();
+          orig_offset = aligned_offset + current_size;
         }
 
         {
@@ -201,8 +216,8 @@ IOStatus RandomAccessFileReader::Read(const IOOptions& opts, uint64_t offset,
           // one iteration of this loop, so we don't need to check and adjust
           // the opts.timeout before calling file_->Read
           assert(!opts.timeout.count() || allowed == read_size);
-          io_s = file_->Read(aligned_offset + buf.CurrentSize(), allowed, opts,
-                             &tmp, buf.Destination(), dbg);
+          io_s = file_->Read(aligned_offset + current_size, allowed, opts, &tmp,
+                             aligned_scratch + current_size, dbg);
         }
         if (ShouldNotifyListeners()) {
           auto finish_ts = FileOperationInfo::FinishNow();
@@ -214,19 +229,22 @@ IOStatus RandomAccessFileReader::Read(const IOOptions& opts, uint64_t offset,
           }
         }
 
-        buf.Size(buf.CurrentSize() + tmp.size());
+        current_size += tmp.size();
+        if (direct_io_buffer == nullptr) {
+          buf.Size(current_size);
+        }
         if (!io_s.ok() || tmp.size() < allowed) {
           break;
         }
       }
       size_t res_len = 0;
-      if (io_s.ok() && offset_advance < buf.CurrentSize()) {
-        res_len = std::min(buf.CurrentSize() - offset_advance, n);
-        if (aligned_buf == nullptr) {
-          buf.Read(scratch, offset_advance, res_len);
+      if (io_s.ok() && offset_advance < current_size) {
+        res_len = std::min(current_size - offset_advance, n);
+        if (direct_io_buffer != nullptr) {
+          scratch = aligned_scratch + offset_advance;
         } else {
-          scratch = buf.BufferStart() + offset_advance;
-          *aligned_buf = buf.Release();
+          assert(scratch != nullptr);
+          buf.Read(scratch, offset_advance, res_len);
         }
       }
       *result = Slice(scratch, res_len);
@@ -335,13 +353,18 @@ bool TryMerge(FSReadRequest* dest, const FSReadRequest& src) {
   return true;
 }
 
-IOStatus RandomAccessFileReader::MultiRead(const IOOptions& opts,
-                                           FSReadRequest* read_reqs,
-                                           size_t num_reqs,
-                                           AlignedBuf* aligned_buf,
-                                           IODebugContext* dbg) const {
-  (void)aligned_buf;  // suppress warning of unused variable in LITE mode
+IOStatus RandomAccessFileReader::MultiRead(
+    const IOOptions& opts, FSReadRequest* read_reqs, size_t num_reqs,
+    AlignedBufferAllocationContext* direct_io_buffer_context,
+    IODebugContext* dbg) const {
   assert(num_reqs > 0);
+  AlignedBuffer* direct_io_buffer = direct_io_buffer_context != nullptr
+                                        ? direct_io_buffer_context->buffer
+                                        : nullptr;
+  const AlignedBuffer::Allocator* direct_io_allocator =
+      direct_io_buffer_context != nullptr ? direct_io_buffer_context->allocator
+                                          : nullptr;
+  assert(direct_io_buffer != nullptr);
 
 #ifndef NDEBUG
   for (size_t i = 0; i < num_reqs - 1; ++i) {
@@ -403,16 +426,20 @@ IOStatus RandomAccessFileReader::MultiRead(const IOOptions& opts,
       for (const auto& r : aligned_reqs) {
         total_len += r.len;
       }
-      AlignedBuffer buf;
-      buf.Alignment(alignment);
-      buf.AllocateNewBuffer(total_len);
-      char* scratch = buf.BufferStart();
-      for (auto& r : aligned_reqs) {
-        r.scratch = scratch;
-        scratch += r.len;
+      direct_io_buffer->Alignment(alignment);
+      Status allocate_status =
+          direct_io_buffer->AllocateNewBuffer(total_len, direct_io_allocator);
+      if (!allocate_status.ok()) {
+        io_s = status_to_io_status(std::move(allocate_status));
+      }
+      if (io_s.ok()) {
+        char* scratch = direct_io_buffer->BufferStart();
+        for (auto& r : aligned_reqs) {
+          r.scratch = scratch;
+          scratch += r.len;
+        }
       }
 
-      *aligned_buf = buf.Release();
       fs_reqs = aligned_reqs.data();
       num_fs_reqs = aligned_reqs.size();
     }
@@ -422,7 +449,7 @@ IOStatus RandomAccessFileReader::MultiRead(const IOOptions& opts,
       start_ts = FileOperationInfo::StartNow();
     }
 
-    {
+    if (io_s.ok()) {
       IOSTATS_CPU_TIMER_GUARD(cpu_read_nanos, clock_);
       if (rate_limiter_priority != Env::IO_TOTAL && rate_limiter_ != nullptr) {
         // TODO: ideally we should call `RateLimiter::RequestToken()` for
@@ -455,7 +482,7 @@ IOStatus RandomAccessFileReader::MultiRead(const IOOptions& opts,
       RecordInHistogram(stats_, MULTIGET_IO_BATCH_SIZE, num_fs_reqs);
     }
 
-    if (use_direct_io()) {
+    if (use_direct_io() && io_s.ok()) {
       // Populate results in the unaligned read requests.
       size_t aligned_i = 0;
       for (size_t i = 0; i < num_reqs; i++) {
@@ -481,19 +508,20 @@ IOStatus RandomAccessFileReader::MultiRead(const IOOptions& opts,
       }
     }
 
+    const bool overall_io_ok = io_s.ok();
     for (size_t i = 0; i < num_reqs; ++i) {
+      const IOStatus& req_status = overall_io_ok ? read_reqs[i].status : io_s;
+      const size_t result_size = overall_io_ok ? read_reqs[i].result.size() : 0;
       if (ShouldNotifyListeners()) {
         auto finish_ts = FileOperationInfo::FinishNow();
-        NotifyOnFileReadFinish(read_reqs[i].offset, read_reqs[i].result.size(),
-                               start_ts, finish_ts, read_reqs[i].status);
+        NotifyOnFileReadFinish(read_reqs[i].offset, result_size, start_ts,
+                               finish_ts, req_status);
       }
-      if (!read_reqs[i].status.ok()) {
-        NotifyOnIOError(read_reqs[i].status, FileOperationType::kRead,
-                        file_name(), read_reqs[i].result.size(),
-                        read_reqs[i].offset);
+      if (!req_status.ok()) {
+        NotifyOnIOError(req_status, FileOperationType::kRead, file_name(),
+                        result_size, read_reqs[i].offset);
       }
-      RecordIOStats(stats_, file_temperature_, is_last_level_,
-                    read_reqs[i].result.size());
+      RecordIOStats(stats_, file_temperature_, is_last_level_, result_size);
     }
     SetPerfLevel(prev_perf_level);
   }
@@ -517,16 +545,26 @@ IOStatus RandomAccessFileReader::PrepareIOOptions(const ReadOptions& ro,
 
 // Notes for when direct_io is enabled:
 // Unless req.offset, req.len, req.scratch are all already aligned,
-// RandomAccessFileReader will creats aligned requests and aligned buffer for
-// the request. User should only provide either req.scratch or aligned_buf. If
-// only req.scratch is provided, result will be copied from allocated aligned
-// buffer to req.scratch. If only alignd_buf is provided, it will be set to
-// the ailgned buf allocated by RandomAccessFileReader and saves a copy.
+// RandomAccessFileReader creates an aligned request and aligned buffer for the
+// request. If direct_io_buffer_context is provided, its buffer owns the aligned
+// backing storage and its optional allocator is used only for this allocation.
+// Otherwise, callers should provide either req.scratch or aligned_buf. If only
+// req.scratch is provided, the result is copied from the allocated aligned
+// buffer to req.scratch. If only aligned_buf is provided, it is set to the
+// aligned buffer allocated by RandomAccessFileReader and saves a copy.
 IOStatus RandomAccessFileReader::ReadAsync(
     FSReadRequest& req, const IOOptions& opts,
     std::function<void(FSReadRequest&, void*)> cb, void* cb_arg,
     void** io_handle, IOHandleDeleter* del_fn, AlignedBuf* aligned_buf,
-    IODebugContext* dbg) {
+    IODebugContext* dbg,
+    AlignedBufferAllocationContext* direct_io_buffer_context) {
+  AlignedBuffer* direct_io_buffer = direct_io_buffer_context != nullptr
+                                        ? direct_io_buffer_context->buffer
+                                        : nullptr;
+  const AlignedBuffer::Allocator* direct_io_allocator =
+      direct_io_buffer_context != nullptr ? direct_io_buffer_context->allocator
+                                          : nullptr;
+  assert(direct_io_buffer_context == nullptr || direct_io_buffer != nullptr);
   IOStatus s;
   TEST_SYNC_POINT_CALLBACK("RandomAccessFileReader::ReadAsync:InjectStatus",
                            &s);
@@ -546,7 +584,8 @@ IOStatus RandomAccessFileReader::ReadAsync(
   }
 
   size_t alignment = file_->GetRequiredBufferAlignment();
-  bool is_aligned = (req.offset & (alignment - 1)) == 0 &&
+  bool is_aligned = direct_io_buffer == nullptr &&
+                    (req.offset & (alignment - 1)) == 0 &&
                     (req.len & (alignment - 1)) == 0 &&
                     (uintptr_t(req.scratch) & (alignment - 1)) == 0;
   read_async_info->is_aligned_ = is_aligned;
@@ -556,21 +595,27 @@ IOStatus RandomAccessFileReader::ReadAsync(
     FSReadRequest aligned_req = Align(req, alignment);
     aligned_req.status.PermitUncheckedError();
 
-    // Allocate aligned buffer.
-    read_async_info->buf_.Alignment(alignment);
-    read_async_info->buf_.AllocateNewBuffer(aligned_req.len);
-
-    // Set rem fields in aligned FSReadRequest.
-    aligned_req.scratch = read_async_info->buf_.BufferStart();
+    AlignedBuffer* aligned_read_buffer =
+        direct_io_buffer != nullptr ? direct_io_buffer : &read_async_info->buf_;
+    aligned_read_buffer->Alignment(alignment);
+    Status allocate_status = aligned_read_buffer->AllocateNewBuffer(
+        aligned_req.len, direct_io_allocator);
+    if (!allocate_status.ok()) {
+      delete read_async_info;
+      return status_to_io_status(std::move(allocate_status));
+    }
+    aligned_req.scratch = aligned_read_buffer->BufferStart();
 
     // Set user provided fields to populate back in callback.
     read_async_info->user_scratch_ = req.scratch;
     read_async_info->user_aligned_buf_ = aligned_buf;
+    read_async_info->direct_io_buffer_ = direct_io_buffer;
     read_async_info->user_len_ = req.len;
     read_async_info->user_offset_ = req.offset;
     read_async_info->user_result_ = req.result;
 
-    assert(read_async_info->buf_.CurrentSize() == 0);
+    assert(direct_io_buffer != nullptr ||
+           read_async_info->buf_.CurrentSize() == 0);
 
     StopWatch sw(clock_, stats_, hist_type_,
                  GetFileReadHistograms(stats_, opts.io_activity),
@@ -618,20 +663,25 @@ void RandomAccessFileReader::ReadAsyncCallback(FSReadRequest& req,
     user_req.result = req.result;
     user_req.status = req.status;
 
-    read_async_info->buf_.Size(read_async_info->buf_.CurrentSize() +
-                               req.result.size());
+    size_t current_size = req.result.size();
+    if (read_async_info->direct_io_buffer_ == nullptr) {
+      read_async_info->buf_.Size(read_async_info->buf_.CurrentSize() +
+                                 req.result.size());
+      current_size = read_async_info->buf_.CurrentSize();
+    }
 
     size_t offset_advance_len = static_cast<size_t>(
         /*offset_passed_by_user=*/read_async_info->user_offset_ -
         /*aligned_offset=*/req.offset);
 
     size_t res_len = 0;
-    if (req.status.ok() &&
-        offset_advance_len < read_async_info->buf_.CurrentSize()) {
-      res_len =
-          std::min(read_async_info->buf_.CurrentSize() - offset_advance_len,
-                   read_async_info->user_len_);
-      if (read_async_info->user_aligned_buf_ == nullptr) {
+    if (req.status.ok() && offset_advance_len < current_size) {
+      res_len = std::min(current_size - offset_advance_len,
+                         read_async_info->user_len_);
+      if (read_async_info->direct_io_buffer_ != nullptr) {
+        user_req.scratch = read_async_info->direct_io_buffer_->BufferStart() +
+                           offset_advance_len;
+      } else if (read_async_info->user_aligned_buf_ == nullptr) {
         // Copy the data into user's scratch.
 // Clang analyzer assumes that it will take use_direct_io() == false in
 // ReadAsync and use_direct_io() == true in Callback which cannot be true.

file/random_access_file_reader.h

@@ -9,6 +9,7 @@
 
 #pragma once
 #include <atomic>
+#include <functional>
 #include <sstream>
 #include <string>
 
@@ -27,6 +28,15 @@ class SystemClock;
 
 using AlignedBuf = FSAllocationPtr;
 
+struct AlignedBufferAllocationContext {
+  // `allocator` is intentionally not owned. RandomAccessFileReader invokes it
+  // only while allocating `buffer`, before Read/ReadAsync/MultiRead returns.
+  // For ReadAsync, `buffer` is still referenced by the completion callback and
+  // must outlive the async operation.
+  AlignedBuffer* buffer = nullptr;
+  const AlignedBuffer::Allocator* allocator = nullptr;
+};
+
 // Align the request r according to alignment and return the aligned result.
 FSReadRequest Align(const FSReadRequest& r, size_t alignment);
 
@@ -97,6 +107,7 @@ class RandomAccessFileReader {
           start_time_(start_time),
           user_scratch_(nullptr),
           user_aligned_buf_(nullptr),
+          direct_io_buffer_(nullptr),
           user_offset_(0),
           user_len_(0),
           is_aligned_(false) {}
@@ -108,6 +119,10 @@ class RandomAccessFileReader {
     // Below fields stores the parameters passed by caller in case of direct_io.
     char* user_scratch_;
     AlignedBuf* user_aligned_buf_;
+    // Raw pointer to caller-owned direct-I/O storage used when forming the
+    // user-visible result in ReadAsyncCallback. The caller must keep it alive
+    // until the async operation completes or is cancelled.
+    AlignedBuffer* direct_io_buffer_;
     uint64_t user_offset_;
     size_t user_len_;
     Slice user_result_;
@@ -157,23 +172,28 @@ class RandomAccessFileReader {
   // 1. if using mmap, result is stored in a buffer other than scratch;
   // 2. if not using mmap, result is stored in the buffer starting from scratch.
   //
-  // In direct IO mode, an aligned buffer is allocated internally.
-  // 1. If aligned_buf is null, then results are copied to the buffer
-  // starting from scratch;
-  // 2. Otherwise, scratch is not used and can be null, the aligned_buf owns
-  // the internally allocated buffer on return, and the result refers to a
-  // region in aligned_buf.
-  IOStatus Read(const IOOptions& opts, uint64_t offset, size_t n, Slice* result,
-                char* scratch, AlignedBuf* aligned_buf,
-                IODebugContext* dbg = nullptr) const;
+  // In direct IO mode, if direct_io_buffer_context is provided then it
+  // allocates the aligned buffer and the result refers to a region in
+  // direct_io_buffer_context->buffer. Otherwise, results are returned in
+  // scratch; unaligned reads use an internal aligned buffer and copy the
+  // requested subrange to scratch.
+  IOStatus Read(
+      const IOOptions& opts, uint64_t offset, size_t n, Slice* result,
+      char* scratch,
+      AlignedBufferAllocationContext* direct_io_buffer_context = nullptr,
+      IODebugContext* dbg = nullptr) const;
 
   // REQUIRES:
   // num_reqs > 0, reqs do not overlap, and offsets in reqs are increasing.
-  // In non-direct IO mode, aligned_buf should be null;
-  // In direct IO mode, aligned_buf stores the aligned buffer allocated inside
-  // MultiRead, the result Slices in reqs refer to aligned_buf.
+  // MultiRead uses direct_io_buffer_context to allocate the aligned buffer in
+  // direct IO mode. The result Slices in reqs refer to
+  // direct_io_buffer_context->buffer, so callers must keep it alive while those
+  // Slices are used. Callers should pass a default-constructed AlignedBuffer
+  // when default heap-backed allocation is sufficient. direct_io_buffer_context
+  // is ignored in non-direct IO mode.
   IOStatus MultiRead(const IOOptions& opts, FSReadRequest* reqs,
-                     size_t num_reqs, AlignedBuf* aligned_buf,
+                     size_t num_reqs,
+                     AlignedBufferAllocationContext* direct_io_buffer_context,
                      IODebugContext* dbg = nullptr) const;
 
   IOStatus Prefetch(const IOOptions& opts, uint64_t offset, size_t n,
@@ -190,10 +210,12 @@ class RandomAccessFileReader {
   IOStatus PrepareIOOptions(const ReadOptions& ro, IOOptions& opts,
                             IODebugContext* dbg = nullptr) const;
 
-  IOStatus ReadAsync(FSReadRequest& req, const IOOptions& opts,
-                     std::function<void(FSReadRequest&, void*)> cb,
-                     void* cb_arg, void** io_handle, IOHandleDeleter* del_fn,
-                     AlignedBuf* aligned_buf, IODebugContext* dbg = nullptr);
+  IOStatus ReadAsync(
+      FSReadRequest& req, const IOOptions& opts,
+      std::function<void(FSReadRequest&, void*)> cb, void* cb_arg,
+      void** io_handle, IOHandleDeleter* del_fn, AlignedBuf* aligned_buf,
+      IODebugContext* dbg = nullptr,
+      AlignedBufferAllocationContext* direct_io_buffer_context = nullptr);
 
   void ReadAsyncCallback(FSReadRequest& req, void* cb_arg);
 };

file/random_access_file_reader_test.cc

@@ -81,15 +81,91 @@ TEST_F(RandomAccessFileReaderTest, ReadDirectIO) {
   size_t offset = page_size / 2;
   size_t len = page_size / 3;
   Slice result;
-  AlignedBuf buf;
   for (Env::IOPriority rate_limiter_priority : {Env::IO_LOW, Env::IO_TOTAL}) {
     IOOptions io_opts;
     io_opts.rate_limiter_priority = rate_limiter_priority;
-    ASSERT_OK(r->Read(io_opts, offset, len, &result, nullptr, &buf));
+    AlignedBuffer direct_io_buffer;
+    AlignedBufferAllocationContext direct_io_context{&direct_io_buffer};
+    ASSERT_OK(r->Read(io_opts, offset, len, &result, nullptr,
+                      &direct_io_context,
+                      /*dbg=*/nullptr));
     ASSERT_EQ(result.ToString(), content.substr(offset, len));
   }
 }
 
+TEST_F(RandomAccessFileReaderTest, ReadDirectIOCopiesToScratch) {
+  std::string fname = "read-direct-io-copies-to-scratch";
+  Random rand(0);
+  std::string content = rand.RandomString(kDefaultPageSize);
+  Write(fname, content);
+
+  FileOptions opts;
+  opts.use_direct_reads = true;
+  std::unique_ptr<RandomAccessFileReader> r;
+  Read(fname, opts, &r);
+  ASSERT_TRUE(r->use_direct_io());
+
+  const size_t page_size = r->file()->GetRequiredBufferAlignment();
+  size_t offset = page_size / 2;
+  size_t len = page_size / 3;
+  std::string scratch(len, '\0');
+  Slice result;
+  ASSERT_OK(r->Read(IOOptions(), offset, len, &result, scratch.data(),
+                    /*direct_io_buffer=*/nullptr, /*dbg=*/nullptr));
+  ASSERT_EQ(result.data(), scratch.data());
+  ASSERT_EQ(result.ToString(), content.substr(offset, len));
+}
+
+TEST_F(RandomAccessFileReaderTest, ReadDirectIOUsesExternalBuffer) {
+  std::string fname = "read-direct-io-external-buffer";
+  Random rand(0);
+  std::string content = rand.RandomString(kDefaultPageSize);
+  Write(fname, content);
+
+  FileOptions opts;
+  opts.use_direct_reads = true;
+  std::unique_ptr<RandomAccessFileReader> r;
+  Read(fname, opts, &r);
+  ASSERT_TRUE(r->use_direct_io());
+
+  const size_t page_size = r->file()->GetRequiredBufferAlignment();
+  const size_t offset = page_size / 4;
+  const size_t len = page_size / 2;
+
+  AlignedBuffer external_storage;
+  int allocations = 0;
+  size_t requested_size = 0;
+  size_t requested_alignment = 0;
+  AlignedBuffer::Allocator allocator =
+      [&](size_t size, size_t alignment,
+          AlignedBuffer::ExternalAllocation* out) {
+        ++allocations;
+        requested_size = size;
+        requested_alignment = alignment;
+        external_storage.Alignment(alignment);
+        external_storage.AllocateNewBuffer(size);
+        out->data = external_storage.BufferStart();
+        out->size = external_storage.Capacity();
+        out->owner =
+            FSAllocationPtr(external_storage.BufferStart(), [](void*) {});
+        return Status::OK();
+      };
+
+  AlignedBuffer direct_io_buffer;
+  AlignedBufferAllocationContext direct_io_context{&direct_io_buffer,
+                                                   &allocator};
+  Slice result;
+  ASSERT_OK(r->Read(IOOptions(), offset, len, &result, /*scratch=*/nullptr,
+                    &direct_io_context, /*dbg=*/nullptr));
+  ASSERT_EQ(result.ToString(), content.substr(offset, len));
+
+  ASSERT_EQ(allocations, 1);
+  ASSERT_EQ(requested_alignment, page_size);
+  ASSERT_EQ(requested_size, page_size);
+  ASSERT_EQ(direct_io_buffer.BufferStart(), external_storage.BufferStart());
+  ASSERT_EQ(result.data(), external_storage.BufferStart() + offset);
+}
+
 TEST_F(RandomAccessFileReaderTest, MultiReadDirectIO) {
   std::vector<FSReadRequest> aligned_reqs;
   ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack(
@@ -146,10 +222,11 @@ TEST_F(RandomAccessFileReaderTest, MultiReadDirectIO) {
     std::vector<FSReadRequest> reqs;
     reqs.push_back(std::move(r0));
     reqs.push_back(std::move(r1));
-    AlignedBuf aligned_buf;
+    AlignedBuffer direct_io_buffer;
+    AlignedBufferAllocationContext direct_io_context{&direct_io_buffer};
     IODebugContext dbg;
-    ASSERT_OK(r->MultiRead(IOOptions(), reqs.data(), reqs.size(), &aligned_buf,
-                           &dbg));
+    ASSERT_OK(r->MultiRead(IOOptions(), reqs.data(), reqs.size(),
+                           &direct_io_context, &dbg));
 
     AssertResult(content, reqs);
 
@@ -192,10 +269,11 @@ TEST_F(RandomAccessFileReaderTest, MultiReadDirectIO) {
     reqs.push_back(std::move(r0));
     reqs.push_back(std::move(r1));
     reqs.push_back(std::move(r2));
-    AlignedBuf aligned_buf;
+    AlignedBuffer direct_io_buffer;
+    AlignedBufferAllocationContext direct_io_context{&direct_io_buffer};
     IODebugContext dbg;
-    ASSERT_OK(r->MultiRead(IOOptions(), reqs.data(), reqs.size(), &aligned_buf,
-                           &dbg));
+    ASSERT_OK(r->MultiRead(IOOptions(), reqs.data(), reqs.size(),
+                           &direct_io_context, &dbg));
 
     AssertResult(content, reqs);
 
@@ -238,10 +316,11 @@ TEST_F(RandomAccessFileReaderTest, MultiReadDirectIO) {
     reqs.push_back(std::move(r0));
     reqs.push_back(std::move(r1));
     reqs.push_back(std::move(r2));
-    AlignedBuf aligned_buf;
+    AlignedBuffer direct_io_buffer;
+    AlignedBufferAllocationContext direct_io_context{&direct_io_buffer};
     IODebugContext dbg;
-    ASSERT_OK(r->MultiRead(IOOptions(), reqs.data(), reqs.size(), &aligned_buf,
-                           &dbg));
+    ASSERT_OK(r->MultiRead(IOOptions(), reqs.data(), reqs.size(),
+                           &direct_io_context, &dbg));
 
     AssertResult(content, reqs);
 
@@ -276,10 +355,11 @@ TEST_F(RandomAccessFileReaderTest, MultiReadDirectIO) {
     std::vector<FSReadRequest> reqs;
     reqs.push_back(std::move(r0));
     reqs.push_back(std::move(r1));
-    AlignedBuf aligned_buf;
+    AlignedBuffer direct_io_buffer;
+    AlignedBufferAllocationContext direct_io_context{&direct_io_buffer};
     IODebugContext dbg;
-    ASSERT_OK(r->MultiRead(IOOptions(), reqs.data(), reqs.size(), &aligned_buf,
-                           &dbg));
+    ASSERT_OK(r->MultiRead(IOOptions(), reqs.data(), reqs.size(),
+                           &direct_io_context, &dbg));
 
     AssertResult(content, reqs);
 
@@ -299,6 +379,71 @@ TEST_F(RandomAccessFileReaderTest, MultiReadDirectIO) {
   ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->ClearAllCallBacks();
 }
 
+TEST_F(RandomAccessFileReaderTest, MultiReadDirectIOUsesExternalBuffer) {
+  std::string fname = "multi-read-direct-io-external-buffer";
+  Random rand(0);
+  std::string content = rand.RandomString(3 * kDefaultPageSize);
+  Write(fname, content);
+
+  FileOptions opts;
+  opts.use_direct_reads = true;
+  std::unique_ptr<RandomAccessFileReader> r;
+  Read(fname, opts, &r);
+  ASSERT_TRUE(r->use_direct_io());
+
+  const size_t page_size = r->file()->GetRequiredBufferAlignment();
+  FSReadRequest r0;
+  r0.offset = page_size / 4;
+  r0.len = page_size / 2;
+  r0.scratch = nullptr;
+
+  FSReadRequest r1;
+  r1.offset = 2 * page_size + page_size / 4;
+  r1.len = page_size / 2;
+  r1.scratch = nullptr;
+
+  std::vector<FSReadRequest> reqs;
+  reqs.push_back(std::move(r0));
+  reqs.push_back(std::move(r1));
+
+  AlignedBuffer external_storage;
+  int allocations = 0;
+  size_t requested_size = 0;
+  size_t requested_alignment = 0;
+  AlignedBuffer::Allocator allocator =
+      [&](size_t size, size_t alignment,
+          AlignedBuffer::ExternalAllocation* out) {
+        ++allocations;
+        requested_size = size;
+        requested_alignment = alignment;
+        external_storage.Alignment(alignment);
+        external_storage.AllocateNewBuffer(size);
+        out->data = external_storage.BufferStart();
+        out->size = external_storage.Capacity();
+        out->owner =
+            FSAllocationPtr(external_storage.BufferStart(), [](void*) {});
+        return Status::OK();
+      };
+
+  AlignedBuffer direct_io_buffer;
+  AlignedBufferAllocationContext direct_io_context{&direct_io_buffer,
+                                                   &allocator};
+  ASSERT_OK(r->MultiRead(IOOptions(), reqs.data(), reqs.size(),
+                         &direct_io_context, /*dbg=*/nullptr));
+  AssertResult(content, reqs);
+
+  ASSERT_EQ(allocations, 1);
+  ASSERT_EQ(requested_alignment, page_size);
+  ASSERT_EQ(requested_size, 2 * page_size);
+  ASSERT_EQ(direct_io_buffer.BufferStart(), external_storage.BufferStart());
+  const char* storage_begin = external_storage.BufferStart();
+  const char* storage_end = storage_begin + external_storage.Capacity();
+  for (const auto& req : reqs) {
+    ASSERT_GE(req.result.data(), storage_begin);
+    ASSERT_LE(req.result.data() + req.result.size(), storage_end);
+  }
+}
+
 TEST(FSReadRequest, Align) {
   FSReadRequest r;
   r.offset = 2000;

include/rocksdb/cleanable.h

@@ -107,6 +107,7 @@ class SharedCleanablePtr {
   Cleanable* operator->();
   // Get as raw pointer to Cleanable
   Cleanable* get();
+  const Cleanable* get() const;
 
   // Creates a (virtual) copy of this SharedCleanablePtr and registers its
   // destruction with target, so that the cleanups registered with the

include/rocksdb/io_dispatcher.h

@@ -260,6 +260,17 @@ class ReadSet {
   Status PollAndProcessAsyncIO(
       const std::shared_ptr<AsyncIOState>& async_state);
 
+  // Release memory budget acquired for a prefetched block.
+  void ReleasePrefetchMemory(size_t block_index);
+
+  // Stop tracking one block from an in-flight async IO request. If this was
+  // the last block using the request, abort and delete the IO handle before the
+  // async state is released.
+  void ReleaseAsyncIOForBlock(size_t block_index);
+
+  // Delete a completed or aborted async IO handle exactly once.
+  void DeleteAsyncIOHandle(const std::shared_ptr<AsyncIOState>& async_state);
+
   // Perform synchronous read for a specific block
   Status SyncRead(size_t block_index);
 

include/rocksdb/options.h

@@ -52,6 +52,7 @@ class MergeOperator;
 class Snapshot;
 class MemTableRepFactory;
 class RateLimiter;
+class ReadScopedBlockBufferProvider;
 class Slice;
 class Statistics;
 class InternalKeyComparator;
@@ -2386,6 +2387,30 @@ struct ReadOptions {
   // reads.
   const UserDefinedIndexFactory* table_index_factory = nullptr;
 
+  // EXPERIMENTAL: Optional non-owning provider for data-block storage pinned by
+  // scans using this ReadOptions. Applications that set it are attempting
+  // advanced performance optimizations and are responsible for ensuring the
+  // provider outlives the iterator/read scope and any provider-backed data that
+  // can remain pinned by that scope.
+  //
+  // This is a raw pointer rather than a shared_ptr because ReadOptions is
+  // copied through stack frames and iterator internals; a shared_ptr would add
+  // refcount overhead to those copies. An internal shadow ReadOptions that
+  // strips ownership would add maintenance overhead for this advanced option.
+  //
+  // Current support is limited to block-based table iterators and MultiScan
+  // data-block reads, and is ignored when mmap reads are enabled. When set,
+  // supported scan reads bypass the data-block cache and use provider-backed
+  // final data-block memory. RocksDB may still use ordinary temporary scratch
+  // for serialized block bytes, such as when a block may be compressed. When
+  // unset, scans use the normal RocksDB data-block backing for the table: use
+  // the configured block cache when present, otherwise use RocksDB-owned block
+  // memory. Index/filter blocks keep their normal block-cache behavior.
+  //
+  // TODO: Extend support to point lookups (Get/MultiGet) once those paths can
+  // preserve provider-backed block ownership.
+  ReadScopedBlockBufferProvider* read_scoped_block_buffer_provider = nullptr;
+
   // *** END options only relevant to iterators or scans ***
 
   // *** BEGIN options for RocksDB internal use only ***

include/rocksdb/table.h

@@ -23,6 +23,7 @@
 #include <unordered_map>
 
 #include "rocksdb/cache.h"
+#include "rocksdb/cleanable.h"
 #include "rocksdb/customizable.h"
 #include "rocksdb/env.h"
 #include "rocksdb/options.h"
@@ -46,6 +47,67 @@ struct ConfigOptions;
 struct EnvOptions;
 class UserDefinedIndexFactory;
 
+// Hook for providing read-scoped storage for data blocks read from SST files.
+// This is configured through C++ options rather than OPTIONS files.
+//
+// Current support is limited to block-based table iterator scans and MultiScan
+// data-block reads. Reads using mmap ignore this provider and use normal
+// RocksDB block backing.
+//
+// The provider backs final data-block contents pinned by the scan. RocksDB may
+// still use ordinary temporary scratch for serialized block bytes, such as when
+// reading a block that may be compressed before decompressing or copying the
+// final data block into provider-backed storage.
+//
+// This is separate from MemoryAllocator because each allocation needs a
+// per-lease cleanup handle that RocksDB can attach to pinned blocks/slices, and
+// direct-I/O reads that use provider-backed read buffers need the requested
+// alignment to be passed to the provider.
+//
+// TODO: Extend support to point lookups (Get/MultiGet) once those paths can
+// preserve provider-backed block ownership.
+//
+// Requirements:
+// - If the same provider instance is shared by multiple concurrently active
+//   readers, `Allocate()` must be safe to call concurrently.
+// - `Lease::data` must point to at least `size` bytes of writable contiguous
+//   memory that remains valid until every copy of `Lease::cleanup` is reset.
+// - `Lease::data` must be aligned to `alignment` bytes, where `alignment` is a
+//   power of two and `1` means no special alignment requirement.
+// - When `alignment` is greater than 1, `Lease::size` must also be a multiple
+//   of `alignment` so it can be used as direct-I/O backing storage.
+// - `Lease::cleanup` must be non-null on success. Its cleanup may run on any
+//   thread that releases the last RocksDB reference. If the cleanup touches
+//   provider or other shared state, it must synchronize with Allocate() and
+//   other provider cleanup callbacks.
+// - After `Allocate()` succeeds, RocksDB releases `Lease::cleanup` on all
+//   paths, including later I/O or decompression failure.
+class ReadScopedBlockBufferProvider {
+ public:
+  // A Lease hands one writable backing allocation from the provider to
+  // RocksDB. For a provider-backed block, the final BlockContents data points
+  // into this allocation and RocksDB attaches `cleanup` to the resulting Blocks
+  // and any slices pinned from them. File-read scratch, copying, and
+  // decompression choices are implementation details outside this contract.
+  //
+  // The provider controls allocation reclamation. RocksDB keeps the data valid
+  // by copying `cleanup`; the provider must not reuse or release `data` until
+  // every copy of `cleanup` has been reset. `size` is the usable backing
+  // allocation size and may be larger than the requested allocation size. For
+  // direct-I/O reads, `size` must be a multiple of the requested alignment.
+  struct Lease {
+    // Writable contiguous memory for the loaded block.
+    char* data = nullptr;
+    size_t size = 0;
+    // Reclaims `data` after all derived pinned key/value slices are released.
+    SharedCleanablePtr cleanup;
+  };
+
+  virtual ~ReadScopedBlockBufferProvider() = default;
+
+  virtual Status Allocate(size_t size, size_t alignment, Lease* out) = 0;
+};
+
 // Types of checksums to use for checking integrity of logical blocks within
 // files. All checksums currently use 32 bits of checking power (1 in 4B
 // chance of failing to detect random corruption). Traditionally, the actual

include/rocksdb/version.h

@@ -13,7 +13,7 @@
 // minor or major version number planned for release.
 #define ROCKSDB_MAJOR 11
 #define ROCKSDB_MINOR 4
-#define ROCKSDB_PATCH 0
+#define ROCKSDB_PATCH 3
 
 // Make it easy to do conditional compilation based on version checks, i.e.
 // #if ROCKSDB_VERSION_GE(4, 5, 6)

table/block_based/block_based_table_reader.cc

@@ -87,6 +87,125 @@ CacheAllocationPtr CopyBufferToHeap(MemoryAllocator* allocator, Slice& buf) {
 }
 }  // namespace
 
+Status AllocateReadScopedBlockBuffer(
+    ReadScopedBlockBufferProvider& block_buffer_provider, size_t requested_size,
+    size_t requested_alignment, ReadScopedBlockBufferProvider::Lease* lease) {
+  assert(lease != nullptr);
+
+  Status s = block_buffer_provider.Allocate(requested_size, requested_alignment,
+                                            lease);
+  if (!s.ok()) {
+    return s;
+  }
+  if (lease->data == nullptr || lease->size < requested_size ||
+      lease->cleanup.get() == nullptr) {
+    return Status::InvalidArgument(
+        "ReadScopedBlockBufferProvider returned an invalid lease");
+  }
+  if (requested_alignment > 1 && (reinterpret_cast<uintptr_t>(lease->data) &
+                                  (requested_alignment - 1)) != 0) {
+    return Status::InvalidArgument(
+        "ReadScopedBlockBufferProvider returned a misaligned lease");
+  }
+  return Status::OK();
+}
+
+Status AllocateReadScopedAlignedBuffer(
+    ReadScopedBlockBufferProvider& block_buffer_provider, size_t size,
+    size_t alignment, ReadScopedBlockBufferProvider::Lease* lease,
+    AlignedBuffer::ExternalAllocation* out) {
+  if (out == nullptr) {
+    return Status::InvalidArgument(
+        "AlignedBuffer allocation output is nullptr");
+  }
+  Status s = AllocateReadScopedBlockBuffer(block_buffer_provider, size,
+                                           alignment, lease);
+  if (!s.ok()) {
+    return s;
+  }
+
+  SharedCleanablePtr owner_cleanup = lease->cleanup;
+  out->data = lease->data;
+  out->size = lease->size;
+  out->owner = FSAllocationPtr(
+      lease->data, [owner_cleanup = std::move(owner_cleanup)](void*) mutable {
+        owner_cleanup.Reset();
+      });
+  return Status::OK();
+}
+
+AlignedBuffer::Allocator MakeReadScopedAlignedBufferAllocator(
+    ReadScopedBlockBufferProviderRef block_buffer_provider,
+    ReadScopedBlockBufferProvider::Lease* lease) {
+  assert(block_buffer_provider.has_value());
+  assert(lease != nullptr);
+  return [block_buffer_provider, lease](
+             size_t size, size_t alignment,
+             AlignedBuffer::ExternalAllocation* out) -> Status {
+    assert(block_buffer_provider.has_value());
+    return AllocateReadScopedAlignedBuffer(block_buffer_provider->get(), size,
+                                           alignment, lease, out);
+  };
+}
+
+ReadScopedBlockBufferProviderRef GetReadScopedBlockBufferProvider(
+    const ReadOptions& ro, bool allow_mmap_reads) {
+  if (allow_mmap_reads || ro.read_scoped_block_buffer_provider == nullptr) {
+    return std::nullopt;
+  }
+  return std::ref(*ro.read_scoped_block_buffer_provider);
+}
+
+bool ShouldUseDataBlockCacheForIterator(
+    const BlockBasedTableOptions& table_options, const ReadOptions& ro,
+    bool allow_mmap_reads) {
+  if (GetReadScopedBlockBufferProvider(ro, allow_mmap_reads).has_value()) {
+    // Provider-backed scan reads use caller-owned read-scope storage as the
+    // data-block backing, so they intentionally skip both lookup and insertion
+    // in the shared data-block cache.
+    return false;
+  }
+  return table_options.block_cache != nullptr;
+}
+
+Status CopyBufferToHeapBlockContents(Slice src, size_t data_size,
+                                     MemoryAllocator* allocator,
+                                     BlockContents* out_contents) {
+  assert(out_contents != nullptr);
+  assert(data_size <= src.size());
+
+  *out_contents = BlockContents(CopyBufferToHeap(allocator, src), data_size);
+#ifndef NDEBUG
+  out_contents->has_trailer = src.size() > data_size;
+#endif
+  return Status::OK();
+}
+
+Status CopyBufferToReadScopedBlockContents(
+    Slice src, size_t data_size,
+    ReadScopedBlockBufferProvider& block_buffer_provider,
+    BlockContents* out_contents) {
+  assert(out_contents != nullptr);
+  assert(data_size <= src.size());
+
+  ReadScopedBlockBufferProvider::Lease read_scoped_lease;
+  Status s = AllocateReadScopedBlockBuffer(block_buffer_provider, src.size(), 1,
+                                           &read_scoped_lease);
+  if (!s.ok()) {
+    return s;
+  }
+
+  memcpy(read_scoped_lease.data, src.data(), src.size());
+  out_contents->data = Slice(read_scoped_lease.data, data_size);
+  out_contents->cleanup = std::move(read_scoped_lease.cleanup);
+  out_contents->backing_size = read_scoped_lease.size;
+  out_contents->AssertSingleOwner();
+#ifndef NDEBUG
+  out_contents->has_trailer = src.size() > data_size;
+#endif
+  return Status::OK();
+}
+
 // Explicitly instantiate templates for each "blocklike" type we use (and
 // before implicit specialization).
 // This makes it possible to keep the template definitions in the .cc file.
@@ -204,7 +323,8 @@ Status ReadAndParseBlockFromFile(
     BlockCreateContext& create_context, bool maybe_compressed,
     UnownedPtr<Decompressor> decomp,
     const PersistentCacheOptions& cache_options,
-    MemoryAllocator* memory_allocator, bool for_compaction, bool async_read) {
+    MemoryAllocator* memory_allocator, bool for_compaction, bool async_read,
+    ReadScopedBlockBufferProviderRef block_buffer_provider = std::nullopt) {
   assert(result);
 
   BlockContents contents;
@@ -212,7 +332,7 @@ Status ReadAndParseBlockFromFile(
       file, prefetch_buffer, footer, options, handle, &contents, ioptions,
       /*do_uncompress*/ maybe_compressed, maybe_compressed,
       TBlocklike::kBlockType, decomp, cache_options, memory_allocator, nullptr,
-      for_compaction);
+      for_compaction, block_buffer_provider);
   Status s;
   // If prefetch_buffer is not allocated, it will fallback to synchronous
   // reading of block contents.
@@ -1715,7 +1835,14 @@ Status BlockBasedTable::LookupAndPinBlocksInCache(
   BlockCacheInterface<TBlocklike> block_cache{
       rep_->table_options.block_cache.get()};
 
-  assert(block_cache);
+  TEST_SYNC_POINT("BlockBasedTable::LookupAndPinBlocksInCache:Start");
+
+  if (!block_cache) {
+    // Callers can reach here after data-block cache use has been disabled by
+    // options such as read-scoped provider-backed scans. Treat the lookup as a
+    // cache miss rather than requiring every caller to special-case null cache.
+    return Status::OK();
+  }
 
   Status s;
   CachableEntry<DecompressorDict> cached_dict;
@@ -2096,9 +2223,8 @@ WithBlocklikeCheck<Status, TBlocklike> BlockBasedTable::RetrieveBlock(
   assert(out_parsed_block);
   assert(out_parsed_block->IsEmpty());
 
-  Status s;
   if (use_cache) {
-    s = MaybeReadBlockAndLoadToCache(
+    Status s = MaybeReadBlockAndLoadToCache(
         prefetch_buffer, ro, handle, decomp, for_compaction, out_parsed_block,
         get_context, lookup_context,
         /*contents=*/nullptr, async_read, use_block_cache_for_lookup);
@@ -2124,16 +2250,22 @@ WithBlocklikeCheck<Status, TBlocklike> BlockBasedTable::RetrieveBlock(
   const bool maybe_compressed =
       BlockTypeMaybeCompressed(TBlocklike::kBlockType) && rep_->decompressor;
   std::unique_ptr<TBlocklike> block;
+  Status s;
 
   {
     Histograms histogram =
         for_compaction ? READ_BLOCK_COMPACTION_MICROS : READ_BLOCK_GET_MICROS;
     StopWatch sw(rep_->ioptions.clock, rep_->ioptions.stats, histogram);
+    ReadScopedBlockBufferProviderRef block_buffer_provider =
+        (!use_cache && TBlocklike::kBlockType == BlockType::kData)
+            ? GetReadScopedBlockBufferProvider(ro,
+                                               rep_->ioptions.allow_mmap_reads)
+            : std::nullopt;
     s = ReadAndParseBlockFromFile(
         rep_->file.get(), prefetch_buffer, rep_->footer, ro, handle, &block,
         rep_->ioptions, rep_->create_context, maybe_compressed, decomp,
         rep_->persistent_cache_options, GetMemoryAllocator(rep_->table_options),
-        for_compaction, async_read);
+        for_compaction, async_read, block_buffer_provider);
 
     if (get_context) {
       switch (TBlocklike::kBlockType) {
@@ -3388,9 +3520,9 @@ void BlockBasedTable::DumpBlockChecksumInfo(const BlockHandle& block_handle,
   IODebugContext dbg;
   IOStatus io_s = rep_->file->PrepareIOOptions(read_options, opts, &dbg);
   if (io_s.ok()) {
-    io_s = rep_->file->Read(opts, block_handle.offset(),
-                            block_size_with_trailer, &raw_block_slice,
-                            raw_block.get(), /*aligned_buf=*/nullptr, &dbg);
+    io_s =
+        rep_->file->Read(opts, block_handle.offset(), block_size_with_trailer,
+                         &raw_block_slice, raw_block.get(), nullptr, &dbg);
   }
   if (io_s.ok() && raw_block_slice.size() == block_size_with_trailer) {
     const char* data = raw_block_slice.data();

table/block_based/block_based_table_reader.h

@@ -33,6 +33,7 @@
 #include "table/table_reader.h"
 #include "table/two_level_iterator.h"
 #include "trace_replay/block_cache_tracer.h"
+#include "util/aligned_buffer.h"
 #include "util/atomic.h"
 #include "util/cast_util.h"
 #include "util/coro_utils.h"
@@ -57,6 +58,55 @@ class GetContext;
 
 using KVPairBlock = std::vector<std::pair<std::string, std::string>>;
 
+// Calls the provider and checks that a successful allocation returned a lease
+// satisfying RocksDB's size, alignment, data pointer, and cleanup contract.
+Status AllocateReadScopedBlockBuffer(
+    ReadScopedBlockBufferProvider& block_buffer_provider, size_t requested_size,
+    size_t requested_alignment, ReadScopedBlockBufferProvider::Lease* lease);
+
+// Allocates a read-scoped provider lease and exposes it as an AlignedBuffer
+// external allocation. The AlignedBuffer owner keeps a cleanup reference so the
+// provider allocation remains valid while direct I/O can write into it.
+Status AllocateReadScopedAlignedBuffer(
+    ReadScopedBlockBufferProvider& block_buffer_provider, size_t size,
+    size_t alignment, ReadScopedBlockBufferProvider::Lease* lease,
+    AlignedBuffer::ExternalAllocation* out);
+
+// Returns an allocator whose direct-I/O allocation comes from the read-scoped
+// provider. The caller must keep `lease` alive until the file reader
+// synchronously invokes the allocator.
+AlignedBuffer::Allocator MakeReadScopedAlignedBufferAllocator(
+    ReadScopedBlockBufferProviderRef block_buffer_provider,
+    ReadScopedBlockBufferProvider::Lease* lease);
+
+// Resolves the read-scoped provider configured for this read. Returns
+// std::nullopt when no provider is configured, or when mmap reads are enabled
+// because mmap file reads can ignore RocksDB-provided scratch buffers.
+ReadScopedBlockBufferProviderRef GetReadScopedBlockBufferProvider(
+    const ReadOptions& ro, bool allow_mmap_reads);
+
+// Centralizes the data-block cache decision for iterator and MultiScan paths.
+// A read-scoped provider skips data-block cache lookup and insertion because it
+// supplies alternate read-scope backing for supported data-block reads.
+bool ShouldUseDataBlockCacheForIterator(
+    const BlockBasedTableOptions& table_options, const ReadOptions& ro,
+    bool allow_mmap_reads);
+
+// Copies block bytes into RocksDB-owned heap storage and records that ownership
+// in BlockContents. `src` may include a block trailer while `data_size` is the
+// payload size.
+Status CopyBufferToHeapBlockContents(Slice src, size_t data_size,
+                                     MemoryAllocator* allocator,
+                                     BlockContents* out_contents);
+
+// Copies block bytes into read-scoped provider storage and records that
+// cleanup in BlockContents. `src` may include a block trailer while
+// `data_size` is the payload size.
+Status CopyBufferToReadScopedBlockContents(
+    Slice src, size_t data_size,
+    ReadScopedBlockBufferProvider& block_buffer_provider,
+    BlockContents* out_contents);
+
 // Reader class for BlockBasedTable format.
 // For the format of BlockBasedTable refer to
 // https://github.com/facebook/rocksdb/wiki/Rocksdb-BlockBasedTable-Format.

table/block_based/block_based_table_reader_impl.h

@@ -90,10 +90,14 @@ TBlockIter* BlockBasedTable::NewDataBlockIterator(
                         lookup_context, for_compaction, /* use_cache */ true,
                         async_read, use_block_cache_for_lookup);
     } else {
-      s = RetrieveBlock(
-          prefetch_buffer, ro, handle, decomp, &block.As<IterBlocklike>(),
-          get_context, lookup_context, for_compaction,
-          /* use_cache */ true, async_read, use_block_cache_for_lookup);
+      const bool use_cache =
+          block_type != BlockType::kData ||
+          ShouldUseDataBlockCacheForIterator(rep_->table_options, ro,
+                                             rep_->ioptions.allow_mmap_reads);
+      s = RetrieveBlock(prefetch_buffer, ro, handle, decomp,
+                        &block.As<IterBlocklike>(), get_context, lookup_context,
+                        for_compaction, use_cache, async_read,
+                        use_block_cache_for_lookup && use_cache);
     }
   }
 

table/block_based/block_based_table_reader_sync_and_async.h

@@ -4,6 +4,7 @@
 //  COPYING file in the root directory) and Apache 2.0 License
 //  (found in the LICENSE.Apache file in the root directory).
 
+#include "util/aligned_buffer.h"
 #include "util/async_file_reader.h"
 #include "util/coro_utils.h"
 
@@ -134,7 +135,8 @@ DEFINE_SYNC_AND_ASYNC(void, BlockBasedTable::RetrieveMultipleBlocks)
     read_reqs.emplace_back(std::move(req));
   }
 
-  AlignedBuf direct_io_buf;
+  AlignedBuffer direct_io_buffer;
+  AlignedBufferAllocationContext direct_io_context{&direct_io_buffer};
   {
     IOOptions opts;
     IODebugContext dbg;
@@ -144,11 +146,11 @@ DEFINE_SYNC_AND_ASYNC(void, BlockBasedTable::RetrieveMultipleBlocks)
       if (file->use_direct_io()) {
 #endif  // WITH_COROUTINES
         s = file->MultiRead(opts, &read_reqs[0], read_reqs.size(),
-                            &direct_io_buf, &dbg);
+                            &direct_io_context, &dbg);
 #if defined(WITH_COROUTINES)
       } else {
         co_await batch->context()->reader().MultiReadAsync(
-            file, opts, &read_reqs[0], read_reqs.size(), &direct_io_buf, &dbg);
+            file, opts, &read_reqs[0], read_reqs.size(), &dbg);
       }
 #endif  // WITH_COROUTINES
     }

table/block_fetcher.cc

@@ -24,6 +24,7 @@
 #include "table/block_based/reader_common.h"
 #include "table/format.h"
 #include "table/persistent_cache_helper.h"
+#include "util/aligned_buffer.h"
 #include "util/compression.h"
 #include "util/stop_watch.h"
 
@@ -151,8 +152,17 @@ inline bool BlockFetcher::TryGetSerializedBlockFromPersistentCache() {
 
 inline void BlockFetcher::PrepareBufferForBlockFromFile() {
   // cache miss read from device
-  if ((do_uncompress_ || ioptions_.allow_mmap_reads) &&
-      block_size_with_trailer_ < kDefaultStackBufferSize) {
+  if (block_buffer_provider_.has_value() && !maybe_compressed_) {
+    Status s = AllocateReadScopedBlockBuffer(block_buffer_provider_->get(),
+                                             block_size_with_trailer_, 1,
+                                             &read_scoped_buf_lease_);
+    if (!s.ok()) {
+      io_status_ = status_to_io_status(std::move(s));
+      return;
+    }
+    used_buf_ = read_scoped_buf_lease_.data;
+  } else if ((do_uncompress_ || ioptions_.allow_mmap_reads) &&
+             block_size_with_trailer_ < kDefaultStackBufferSize) {
     // If we've got a small enough chunk of data, read it in to the
     // trivially allocated stack buffer instead of needing a full malloc()
     //
@@ -165,9 +175,11 @@ inline void BlockFetcher::PrepareBufferForBlockFromFile() {
     // stack buffer, the cost of guessing incorrectly here is one extra memcpy.
     //
     // When `do_uncompress_` is true, we expect the uncompression step will
-    // allocate heap memory for the final result. However this expectation will
-    // be wrong if the block turns out to already be uncompressed, which we
-    // won't know for sure until after reading it.
+    // allocate memory for the final result, using the read-scoped provider if
+    // one is configured. However this expectation will be wrong if the block
+    // turns out to already be uncompressed, which we won't know for sure until
+    // after reading it. In that case provider-backed reads copy the block into
+    // provider storage in `GetBlockContents()`.
     //
     // When `ioptions_.allow_mmap_reads` is true, we do not expect the file
     // reader to use the scratch buffer at all, but instead return a pointer
@@ -231,14 +243,29 @@ inline void BlockFetcher::CopyBufferToCompressedBuf() {
 //    is not compressed
 // 3. heap_buf_ if the block is not compressed
 // 4. compressed_buf_ if the block is compressed
-// 5. direct_io_buf_ if direct IO is enabled or
+// 5. direct_io_buffer_ if direct IO is enabled or
 // 6. underlying file_system scratch is used (FSReadRequest.fs_scratch).
 //
-// After - After this method, if the block is compressed, it should be in
-// compressed_buf_ and heap_buf_ points to compressed_buf_, otherwise should be
-// in heap_buf_.
+// After - After this method, compressed blocks should be in compressed_buf_ and
+// heap_buf_ points to compressed_buf_. Uncompressed blocks should be recorded
+// in *contents_ with either heap ownership or read-scoped cleanup ownership.
 inline void BlockFetcher::GetBlockContents() {
-  if (slice_.data() != used_buf_) {
+  if (read_scoped_buf_lease_.cleanup.get() != nullptr &&
+      compression_type() == kNoCompression) {
+    contents_->data = Slice(slice_.data(), block_size_);
+    contents_->cleanup = std::move(read_scoped_buf_lease_.cleanup);
+    contents_->backing_size = read_scoped_buf_lease_.size;
+    contents_->AssertSingleOwner();
+  } else if (block_buffer_provider_.has_value() &&
+             compression_type() == kNoCompression) {
+    Status s = CopyBufferToReadScopedBlockContents(
+        Slice(slice_.data(), block_size_with_trailer_), block_size_,
+        block_buffer_provider_->get(), contents_);
+    if (!s.ok()) {
+      io_status_ = status_to_io_status(std::move(s));
+      return;
+    }
+  } else if (slice_.data() != used_buf_) {
     // the slice content is not the buffer provided
     *contents_ = BlockContents(Slice(slice_.data(), block_size_));
   } else {
@@ -253,7 +280,7 @@ inline void BlockFetcher::GetBlockContents() {
       } else {
         heap_buf_ = std::move(compressed_buf_);
       }
-    } else if (direct_io_buf_.get() != nullptr || use_fs_scratch_) {
+    } else if (direct_io_buffer_.BufferStart() != nullptr || use_fs_scratch_) {
       if (compression_type() == kNoCompression) {
         CopyBufferToHeapBuf();
       } else {
@@ -281,13 +308,21 @@ void BlockFetcher::ReadBlock(bool retry) {
   // Actual file read
   if (io_status_.ok()) {
     if (file_->use_direct_io()) {
+      AlignedBuffer::Allocator direct_io_allocator;
+      if (block_buffer_provider_.has_value() && !maybe_compressed_) {
+        direct_io_allocator = MakeReadScopedAlignedBufferAllocator(
+            block_buffer_provider_, &read_scoped_buf_lease_);
+      }
+      AlignedBufferAllocationContext direct_io_context{
+          &direct_io_buffer_,
+          direct_io_allocator ? &direct_io_allocator : nullptr};
       PERF_TIMER_GUARD(block_read_time);
       PERF_CPU_TIMER_GUARD(
           block_read_cpu_time,
           ioptions_.env ? ioptions_.env->GetSystemClock().get() : nullptr);
       io_status_ =
           file_->Read(opts, handle_.offset(), block_size_with_trailer_, &slice_,
-                      /*scratch=*/nullptr, &direct_io_buf_, &dbg);
+                      /*scratch=*/nullptr, &direct_io_context, &dbg);
       PERF_COUNTER_ADD(block_read_count, 1);
       used_buf_ = const_cast<char*>(slice_.data());
     } else if (use_fs_scratch_) {
@@ -298,8 +333,10 @@ void BlockFetcher::ReadBlock(bool retry) {
       read_req.offset = handle_.offset();
       read_req.len = block_size_with_trailer_;
       read_req.scratch = nullptr;
+      AlignedBuffer direct_io_buffer;
+      AlignedBufferAllocationContext direct_io_context{&direct_io_buffer};
       io_status_ = file_->MultiRead(opts, &read_req, /*num_reqs=*/1,
-                                    /*AlignedBuf* =*/nullptr, &dbg);
+                                    &direct_io_context, &dbg);
       PERF_COUNTER_ADD(block_read_count, 1);
 
       slice_ = Slice(read_req.result.data(), read_req.result.size());
@@ -307,6 +344,9 @@ void BlockFetcher::ReadBlock(bool retry) {
     } else {
       // It allocates/assign used_buf_
       PrepareBufferForBlockFromFile();
+      if (!io_status_.ok()) {
+        return;
+      }
 
       PERF_TIMER_GUARD(block_read_time);
       PERF_CPU_TIMER_GUARD(
@@ -316,7 +356,7 @@ void BlockFetcher::ReadBlock(bool retry) {
       io_status_ =
           file_->Read(opts, handle_.offset(), /*size*/ block_size_with_trailer_,
                       /*result*/ &slice_, /*scratch*/ used_buf_,
-                      /*aligned_buf=*/nullptr, &dbg);
+                      /*direct_io_buffer=*/nullptr, &dbg);
       PERF_COUNTER_ADD(block_read_count, 1);
 #ifndef NDEBUG
       if (slice_.data() == &stack_buf_[0]) {
@@ -380,7 +420,7 @@ void BlockFetcher::ReadBlock(bool retry) {
     }
   } else {
     ReleaseFileSystemProvidedBuffer(&read_req);
-    direct_io_buf_.reset();
+    direct_io_buffer_.Release();
     compressed_buf_.reset();
     heap_buf_.reset();
     used_buf_ = nullptr;
@@ -423,7 +463,8 @@ IOStatus BlockFetcher::ReadBlockContents() {
     slice_.size_ = block_size_;
     decomp_args_.compressed_data = slice_;
     io_status_ = status_to_io_status(DecompressSerializedBlock(
-        decomp_args_, *decompressor_, contents_, ioptions_, memory_allocator_));
+        decomp_args_, *decompressor_, contents_, ioptions_, memory_allocator_,
+        block_buffer_provider_));
 #ifndef NDEBUG
     num_heap_buf_memcpy_++;
 #endif
@@ -477,9 +518,9 @@ IOStatus BlockFetcher::ReadAsyncBlockContents() {
           // Process the compressed block without trailer
           slice_.size_ = block_size_;
           decomp_args_.compressed_data = slice_;
-          io_status_ = status_to_io_status(
-              DecompressSerializedBlock(decomp_args_, *decompressor_, contents_,
-                                        ioptions_, memory_allocator_));
+          io_status_ = status_to_io_status(DecompressSerializedBlock(
+              decomp_args_, *decompressor_, contents_, ioptions_,
+              memory_allocator_, block_buffer_provider_));
 #ifndef NDEBUG
           num_heap_buf_memcpy_++;
 #endif

table/block_fetcher.h

@@ -39,19 +39,18 @@ namespace ROCKSDB_NAMESPACE {
 
 class BlockFetcher {
  public:
-  BlockFetcher(RandomAccessFileReader* file,
-               FilePrefetchBuffer* prefetch_buffer,
-               const Footer& footer /* ref retained */,
-               const ReadOptions& read_options,
-               const BlockHandle& handle /* ref retained */,
-               BlockContents* contents,
-               const ImmutableOptions& ioptions /* ref retained */,
-               bool do_uncompress, bool maybe_compressed, BlockType block_type,
-               UnownedPtr<Decompressor> decompressor,
-               const PersistentCacheOptions& cache_options /* ref retained */,
-               MemoryAllocator* memory_allocator = nullptr,
-               MemoryAllocator* memory_allocator_compressed = nullptr,
-               bool for_compaction = false)
+  BlockFetcher(
+      RandomAccessFileReader* file, FilePrefetchBuffer* prefetch_buffer,
+      const Footer& footer /* ref retained */, const ReadOptions& read_options,
+      const BlockHandle& handle /* ref retained */, BlockContents* contents,
+      const ImmutableOptions& ioptions /* ref retained */, bool do_uncompress,
+      bool maybe_compressed, BlockType block_type,
+      UnownedPtr<Decompressor> decompressor,
+      const PersistentCacheOptions& cache_options /* ref retained */,
+      MemoryAllocator* memory_allocator = nullptr,
+      MemoryAllocator* memory_allocator_compressed = nullptr,
+      bool for_compaction = false,
+      ReadScopedBlockBufferProviderRef block_buffer_provider = std::nullopt)
       : file_(file),
         prefetch_buffer_(prefetch_buffer),
         footer_(footer),
@@ -68,6 +67,8 @@ class BlockFetcher {
         cache_options_(cache_options),
         memory_allocator_(memory_allocator),
         memory_allocator_compressed_(memory_allocator_compressed),
+        block_buffer_provider_(
+            ioptions.allow_mmap_reads ? std::nullopt : block_buffer_provider),
         for_compaction_(for_compaction) {
     io_status_.PermitUncheckedError();  // TODO(AR) can we improve on this?
     if (CheckFSFeatureSupport(ioptions_.fs.get(), FSSupportedOps::kFSBuffer)) {
@@ -129,12 +130,18 @@ class BlockFetcher {
   const PersistentCacheOptions& cache_options_;
   MemoryAllocator* memory_allocator_;
   MemoryAllocator* memory_allocator_compressed_;
+  // Optional provider used when the fetched block should be backed by
+  // read-scoped storage instead of RocksDB heap memory.
+  ReadScopedBlockBufferProviderRef block_buffer_provider_;
   IOStatus io_status_;
   Slice slice_;
   char* used_buf_ = nullptr;
-  AlignedBuf direct_io_buf_;
+  AlignedBuffer direct_io_buffer_;
   CacheAllocationPtr heap_buf_;
   CacheAllocationPtr compressed_buf_;
+  // Provider lease backing `used_buf_`; moved into BlockContents once the block
+  // is parsed so the provider allocation lives as long as the block does.
+  ReadScopedBlockBufferProvider::Lease read_scoped_buf_lease_;
   char stack_buf_[kDefaultStackBufferSize];
   bool got_from_prefetch_buffer_ = false;
   bool for_compaction_ = false;

table/format.cc

@@ -529,7 +529,7 @@ static Status ReadFooterFromFileInternal(
   }
 
   std::array<char, Footer::kMaxEncodedLength + 1> footer_buf;
-  AlignedBuf internal_buf;
+  AlignedBuffer direct_io_buffer;
   Slice footer_input;
   uint64_t read_offset = (expected_file_size > Footer::kMaxEncodedLength)
                              ? expected_file_size - Footer::kMaxEncodedLength
@@ -545,11 +545,12 @@ static Status ReadFooterFromFileInternal(
                                          Footer::kMaxEncodedLength,
                                          &footer_input, nullptr)) {
     if (file->use_direct_io()) {
+      AlignedBufferAllocationContext direct_io_context{&direct_io_buffer};
       s = file->Read(opts, read_offset, Footer::kMaxEncodedLength,
-                     &footer_input, nullptr, &internal_buf);
+                     &footer_input, nullptr, &direct_io_context);
     } else {
       s = file->Read(opts, read_offset, Footer::kMaxEncodedLength,
-                     &footer_input, footer_buf.data(), nullptr);
+                     &footer_input, footer_buf.data());
     }
     if (!s.ok()) {
       return s;
@@ -683,10 +684,11 @@ uint32_t ComputeBuiltinChecksumWithLastByte(ChecksumType type, const char* data,
   }
 }
 
-Status DecompressBlockData(Decompressor::Args& args, Decompressor& decompressor,
-                           BlockContents* out_contents,
-                           const ImmutableOptions& ioptions,
-                           MemoryAllocator* allocator) {
+Status DecompressBlockData(
+    Decompressor::Args& args, Decompressor& decompressor,
+    BlockContents* out_contents, const ImmutableOptions& ioptions,
+    MemoryAllocator* allocator,
+    ReadScopedBlockBufferProviderRef block_buffer_provider) {
   assert(args.compression_type != kNoCompression && "Invalid compression type");
 
   StopWatchNano timer(ioptions.clock,
@@ -696,13 +698,35 @@ Status DecompressBlockData(Decompressor::Args& args, Decompressor& decompressor,
   if (UNLIKELY(!s.ok())) {
     return s;
   }
-  CacheAllocationPtr ubuf = AllocateBlock(args.uncompressed_size, allocator);
-  s = decompressor.DecompressBlock(args, ubuf.get());
-  if (UNLIKELY(!s.ok())) {
-    return s;
-  }
+  if (block_buffer_provider.has_value()) {
+    // Compressed provider path: allocate the final uncompressed block backing
+    // from the provider, then decompress directly into it.
+    ReadScopedBlockBufferProvider::Lease read_scoped_lease;
+    s = AllocateReadScopedBlockBuffer(block_buffer_provider->get(),
+                                      args.uncompressed_size, 1,
+                                      &read_scoped_lease);
+    if (UNLIKELY(!s.ok())) {
+      return s;
+    }
 
-  *out_contents = BlockContents(std::move(ubuf), args.uncompressed_size);
+    s = decompressor.DecompressBlock(args, read_scoped_lease.data);
+    if (UNLIKELY(!s.ok())) {
+      return s;
+    }
+
+    out_contents->data = Slice(read_scoped_lease.data, args.uncompressed_size);
+    out_contents->cleanup = std::move(read_scoped_lease.cleanup);
+    out_contents->backing_size = read_scoped_lease.size;
+  } else {
+    CacheAllocationPtr ubuf = AllocateBlock(args.uncompressed_size, allocator);
+
+    s = decompressor.DecompressBlock(args, ubuf.get());
+    if (UNLIKELY(!s.ok())) {
+      return s;
+    }
+
+    *out_contents = BlockContents(std::move(ubuf), args.uncompressed_size);
+  }
 
   if (ShouldReportDetailedTime(ioptions.env, ioptions.stats)) {
     RecordTimeToHistogram(ioptions.stats, DECOMPRESSION_TIMES_NANOS,
@@ -733,32 +757,42 @@ Status DecompressBlockData(const char* data, size_t size, CompressionType type,
   args.compression_type = type;
   args.working_area = working_area;
   return DecompressBlockData(args, decompressor, out_contents, ioptions,
-                             allocator);
+                             allocator, std::nullopt);
 }
 
-Status DecompressSerializedBlock(const char* data, size_t size,
-                                 CompressionType type,
-                                 Decompressor& decompressor,
-                                 BlockContents* out_contents,
-                                 const ImmutableOptions& ioptions,
-                                 MemoryAllocator* allocator) {
+Status DecompressSerializedBlock(
+    const char* data, size_t size, CompressionType type,
+    Decompressor& decompressor, BlockContents* out_contents,
+    const ImmutableOptions& ioptions, MemoryAllocator* allocator,
+    ReadScopedBlockBufferProviderRef block_buffer_provider) {
   assert(data[size] != kNoCompression);
   assert(data[size] == static_cast<char>(type));
-  return DecompressBlockData(data, size, type, decompressor, out_contents,
-                             ioptions, allocator);
+  Decompressor::Args args;
+  args.compressed_data = Slice(data, size);
+  args.compression_type = type;
+  return DecompressBlockData(args, decompressor, out_contents, ioptions,
+                             allocator, block_buffer_provider);
 }
 
-Status DecompressSerializedBlock(Decompressor::Args& args,
-                                 Decompressor& decompressor,
-                                 BlockContents* out_contents,
-                                 const ImmutableOptions& ioptions,
-                                 MemoryAllocator* allocator) {
+Status DecompressBlockData(Decompressor::Args& args, Decompressor& decompressor,
+                           BlockContents* out_contents,
+                           const ImmutableOptions& ioptions,
+                           MemoryAllocator* allocator) {
+  return DecompressBlockData(args, decompressor, out_contents, ioptions,
+                             allocator, std::nullopt);
+}
+
+Status DecompressSerializedBlock(
+    Decompressor::Args& args, Decompressor& decompressor,
+    BlockContents* out_contents, const ImmutableOptions& ioptions,
+    MemoryAllocator* allocator,
+    ReadScopedBlockBufferProviderRef block_buffer_provider) {
   assert(args.compressed_data.data()[args.compressed_data.size()] !=
          kNoCompression);
   assert(args.compressed_data.data()[args.compressed_data.size()] ==
          static_cast<char>(args.compression_type));
   return DecompressBlockData(args, decompressor, out_contents, ioptions,
-                             allocator);
+                             allocator, block_buffer_provider);
 }
 
 // Replace the contents of db_host_id with the actual hostname, if db_host_id

table/format.h

@@ -10,7 +10,10 @@
 #pragma once
 
 #include <array>
+#include <cassert>
 #include <cstdint>
+#include <functional>
+#include <optional>
 #include <string>
 
 #include "file/file_prefetch_buffer.h"
@@ -19,6 +22,7 @@
 #include "options/cf_options.h"
 #include "port/malloc.h"
 #include "port/port.h"  // noexcept
+#include "rocksdb/cleanable.h"
 #include "rocksdb/slice.h"
 #include "rocksdb/status.h"
 #include "rocksdb/table.h"
@@ -29,6 +33,9 @@ namespace ROCKSDB_NAMESPACE {
 class RandomAccessFile;
 struct ReadOptions;
 
+using ReadScopedBlockBufferProviderRef =
+    std::optional<std::reference_wrapper<ReadScopedBlockBufferProvider>>;
+
 bool ShouldReportDetailedTime(Env* env, Statistics* stats);
 
 // the length of the magic number in bytes.
@@ -404,6 +411,11 @@ struct BlockContents {
   // Points to block payload (without trailer)
   Slice data;
   CacheAllocationPtr allocation;
+  // Optional shared lifetime token for block bytes not owned through
+  // `allocation`, such as bytes backed by a read-scoped provider.
+  SharedCleanablePtr cleanup;
+  // Usable byte size of the provider/external backing allocation.
+  size_t backing_size = 0;
 
 #ifndef NDEBUG
   // Whether there is a known trailer after what is pointed to by `data`.
@@ -411,6 +423,10 @@ struct BlockContents {
   bool has_trailer = false;
 #endif  // NDEBUG
 
+  void AssertSingleOwner() const {
+    assert(allocation.get() == nullptr || cleanup.get() == nullptr);
+  }
+
   BlockContents() {}
 
   // Does not take ownership of the underlying data bytes.
@@ -427,10 +443,14 @@ struct BlockContents {
   }
 
   // Returns whether the object has ownership of the underlying data bytes.
-  bool own_bytes() const { return allocation.get() != nullptr; }
+  bool own_bytes() const {
+    AssertSingleOwner();
+    return allocation.get() != nullptr || cleanup.get() != nullptr;
+  }
 
   // The additional memory space taken by the block data.
   size_t usable_size() const {
+    AssertSingleOwner();
     // FIXME: doesn't account for possible block trailer
     if (allocation.get() != nullptr) {
       auto allocator = allocation.get_deleter().allocator;
@@ -442,12 +462,15 @@ struct BlockContents {
 #else
       return data.size();
 #endif  // ROCKSDB_MALLOC_USABLE_SIZE
+    } else if (cleanup.get() != nullptr) {
+      return backing_size;
     } else {
       return 0;  // no extra memory is occupied by the data
     }
   }
 
   size_t ApproximateMemoryUsage() const {
+    AssertSingleOwner();
     return usable_size() + sizeof(*this);
   }
 
@@ -456,9 +479,12 @@ struct BlockContents {
   BlockContents& operator=(BlockContents&& other) {
     data = std::move(other.data);
     allocation = std::move(other.allocation);
+    cleanup = std::move(other.cleanup);
+    backing_size = other.backing_size;
 #ifndef NDEBUG
     has_trailer = other.has_trailer;
 #endif  // NDEBUG
+    AssertSingleOwner();
     return *this;
   }
 };
@@ -467,18 +493,17 @@ struct BlockContents {
 // must be compressed and include a trailer beyond `size`. A new buffer is
 // allocated with the given allocator (or default) and the uncompressed
 // contents are returned in `out_contents`. Statistics updated.
-Status DecompressSerializedBlock(const char* data, size_t size,
-                                 CompressionType type,
-                                 Decompressor& decompressor,
-                                 BlockContents* out_contents,
-                                 const ImmutableOptions& ioptions,
-                                 MemoryAllocator* allocator = nullptr);
+Status DecompressSerializedBlock(
+    const char* data, size_t size, CompressionType type,
+    Decompressor& decompressor, BlockContents* out_contents,
+    const ImmutableOptions& ioptions, MemoryAllocator* allocator = nullptr,
+    ReadScopedBlockBufferProviderRef block_buffer_provider = std::nullopt);
 
-Status DecompressSerializedBlock(Decompressor::Args& args,
-                                 Decompressor& decompressor,
-                                 BlockContents* out_contents,
-                                 const ImmutableOptions& ioptions,
-                                 MemoryAllocator* allocator = nullptr);
+Status DecompressSerializedBlock(
+    Decompressor::Args& args, Decompressor& decompressor,
+    BlockContents* out_contents, const ImmutableOptions& ioptions,
+    MemoryAllocator* allocator = nullptr,
+    ReadScopedBlockBufferProviderRef block_buffer_provider = std::nullopt);
 
 // This is a variant of DecompressSerializedBlock that does not expect a
 // block trailer beyond `size`. (CompressionType is passed in.)

unreleased_history/behavior_changes/remote_compaction_parse_failure_falls_back_local.md

@@ -1 +0,0 @@
-Remote compaction now falls back to local compaction when the primary cannot deserialize a successful `CompactionServiceResult` before installing any remote output files.

unreleased_history/behavior_changes/string_to_map_self_contained.md

@@ -1 +0,0 @@
-StringToMap (rocksdb/convenience.h) now preserves the outer braces of nested values so each map entry is in self-contained form and can be embedded directly into another `key=value;` string (e.g. SetOptions) without losing inner ';' delimiters. A new symmetric MapToString utility is provided.

unreleased_history/bug_fixes/abort_all_compactions_reschedule.md

@@ -1 +0,0 @@
-Fixed a bug where `AbortAllCompactions()` could leave automatic compaction work unscheduled when aborting before the background worker picked it, causing compaction and write stalls until DB restart.

unreleased_history/bug_fixes/compaction_service_inaccurate_input_record_count

@@ -1 +0,0 @@
-Fix a bug where db had a false positive compaction corruption error due to remote compaction service result with `has_accurate_num_input_records=false` not being serialized. 

unreleased_history/bug_fixes/prepared_txn_rollback_after_write_failure.md

@@ -1 +0,0 @@
-Fixed WritePrepared TransactionDB cleanup after retryable commit or rollback write failures so prepared transactions remain rollbackable instead of leaving unresolved prepared state.

unreleased_history/bug_fixes/proximal_compaction_range_tombstone

@@ -1 +0,0 @@
-Fix a rare corruption bug for tiered compaction that incorrectly moved last level range tombstones into proximal level. This corruption error is surfaced when force_consistency_checks is enabled.

unreleased_history/performance_improvements/manifest_rotation_foreground_relaxation.md

@@ -1 +0,0 @@
-Reduce the likelihood of user-facing metadata operations blocking on MANIFEST rotation when file creation is slow, such as on remote storage. MANIFEST write batches containing foreground edits from `CreateColumnFamily()`, `DropColumnFamily()`, `CreateColumnFamilyWithImport()`, `IngestExternalFile()`, and `DeleteFilesInRanges()` now get a relaxed file size threshold for triggering MANIFEST rotation; background-only MANIFEST write batches, such as compaction and flush, continue to use the normal threshold. This change should make auto-tuning manifest file size more attractive (see `max_manifest_file_size` and `max_manifest_space_amp_pct` options).

unreleased_history/public_api_changes/c_api_memtable_batch_lookup_optimization.md

@@ -1 +0,0 @@
-Added `rocksdb_options_set_memtable_batch_lookup_optimization()` and `rocksdb_options_get_memtable_batch_lookup_optimization()` to the C API, exposing the existing `AdvancedColumnFamilyOptions::memtable_batch_lookup_optimization` field. This allows C API users (and downstream language bindings) to enable the skip-list memtable's batch-lookup optimization for `MultiGet`, which caches the search path between consecutive keys and reduces per-key cost from O(log N) to O(log d) where d is the distance between consecutive keys.

unreleased_history/public_api_changes/c_api_optimize_multiget_for_io.md

@@ -1 +0,0 @@
-Added `rocksdb_readoptions_set_optimize_multiget_for_io()` and `rocksdb_readoptions_get_optimize_multiget_for_io()` to the C API, exposing the existing `ReadOptions::optimize_multiget_for_io` field. This allows C API users (and downstream language bindings) to opt out of the multi-level parallel MultiGet path, which only takes effect when the library is built with `USE_COROUTINES`.

unreleased_history/public_api_changes/c_api_uniform_cv_threshold_and_kAuto.md

@@ -1 +0,0 @@
-Added `rocksdb_block_based_table_index_block_search_type_auto` enum constant and the `rocksdb_block_based_options_set_uniform_cv_threshold()` setter to the C API. The constant exposes `BlockSearchType::kAuto`, and the setter exposes `BlockBasedTableOptions::uniform_cv_threshold`. Both are required for `kAuto` index-block search to take effect from the C API: without setting `uniform_cv_threshold >= 0`, the per-block `is_uniform` footer bit is never written, and `kAuto` falls back to binary search at read time.

unreleased_history/public_api_changes/on_db_shutdown_begin.md

@@ -1 +0,0 @@
-Added `EventListener::OnDBShutdownBegin`, a callback that fires once when a DB begins shutdown, including when RocksDB cleans up a failed `DB::Open()` attempt.

unreleased_history/public_api_changes/perf_context_captured.md

@@ -1 +0,0 @@
-Added a new PerfContext counter `blob_cache_read_byte` for blob cache bytes read.

util/aligned_buffer.h

@@ -10,10 +10,13 @@
 
 #include <algorithm>
 #include <cassert>
+#include <functional>
+#include <utility>
 
 #include "port/malloc.h"
 #include "port/port.h"
 #include "rocksdb/file_system.h"
+#include "rocksdb/status.h"
 namespace ROCKSDB_NAMESPACE {
 
 // This file contains utilities to handle the alignment of pages and buffers.
@@ -56,6 +59,17 @@ inline size_t Rounddown(size_t x, size_t y) { return (x / y) * y; }
 //   buf.AllocateNewBuffer(2*user_requested_buf_size, /*copy_data*/ true,
 //                         copy_offset, copy_len);
 class AlignedBuffer {
+ public:
+  struct ExternalAllocation {
+    char* data = nullptr;
+    size_t size = 0;
+    FSAllocationPtr owner;
+  };
+
+  using Allocator = std::function<Status(size_t size, size_t alignment,
+                                         ExternalAllocation* out)>;
+
+ private:
   size_t alignment_;
   FSAllocationPtr buf_;
   size_t capacity_;
@@ -179,6 +193,55 @@ class AlignedBuffer {
         [](void* p) { delete[] static_cast<char*>(p); });
   }
 
+  // Allocates a fresh buffer, using the external allocator when provided and
+  // RocksDB heap memory otherwise. This overload does not preserve old
+  // contents, and callers must check its returned Status. Heap-only callers
+  // should use the void overload above so CHECK_STATUS builds do not create an
+  // ignored Status.
+  Status AllocateNewBuffer(size_t requested_capacity,
+                           const Allocator* allocator) {
+    if (allocator == nullptr) {
+      AllocateNewBuffer(requested_capacity);
+      return Status::OK();
+    }
+
+    assert(alignment_ > 0);
+    assert((alignment_ & (alignment_ - 1)) == 0);
+
+    const size_t new_capacity = Roundup(requested_capacity, alignment_);
+    ExternalAllocation allocation;
+    Status s = (*allocator)(new_capacity, alignment_, &allocation);
+    if (!s.ok()) {
+      return s;
+    }
+    if (allocation.data == nullptr) {
+      return Status::InvalidArgument(
+          "AlignedBuffer allocator returned null data");
+    }
+    if (allocation.size < new_capacity) {
+      return Status::InvalidArgument(
+          "AlignedBuffer allocator returned short buffer");
+    }
+    if (!isAligned(allocation.data, alignment_)) {
+      return Status::InvalidArgument(
+          "AlignedBuffer allocator returned misaligned buffer");
+    }
+    if (!isAligned(allocation.size, alignment_)) {
+      return Status::InvalidArgument(
+          "AlignedBuffer allocator returned misaligned size");
+    }
+    if (allocation.owner.get() == nullptr) {
+      return Status::InvalidArgument(
+          "AlignedBuffer allocator returned null owner");
+    }
+
+    bufstart_ = allocation.data;
+    capacity_ = allocation.size;
+    cursize_ = 0;
+    buf_ = std::move(allocation.owner);
+    return Status::OK();
+  }
+
   // Append to the buffer.
   //
   // src         : source to copy the data from.

util/async_file_reader.h

@@ -36,10 +36,9 @@ class AsyncFileReader {
                                             const IOOptions& opts,
                                             FSReadRequest* read_reqs,
                                             size_t num_reqs,
-                                            AlignedBuf* aligned_buf,
                                             IODebugContext* dbg) noexcept {
-    return ReadOperation<ReadAwaiter>{*this,    file,        opts, read_reqs,
-                                      num_reqs, aligned_buf, dbg};
+    return ReadOperation<ReadAwaiter>{*this,     file,     opts,
+                                      read_reqs, num_reqs, dbg};
   }
 
  private:
@@ -50,8 +49,7 @@ class AsyncFileReader {
    public:
     explicit ReadAwaiter(AsyncFileReader& reader, RandomAccessFileReader* file,
                          const IOOptions& opts, FSReadRequest* read_reqs,
-                         size_t num_reqs, AlignedBuf* /*aligned_buf*/,
-                         IODebugContext* dbg) noexcept
+                         size_t num_reqs, IODebugContext* dbg) noexcept
         : reader_(reader),
           file_(file),
           opts_(opts),
@@ -105,20 +103,18 @@ class AsyncFileReader {
     explicit ReadOperation(AsyncFileReader& reader,
                            RandomAccessFileReader* file, const IOOptions& opts,
                            FSReadRequest* read_reqs, size_t num_reqs,
-                           AlignedBuf* aligned_buf,
                            IODebugContext* dbg) noexcept
         : reader_(reader),
           file_(file),
           opts_(opts),
           read_reqs_(read_reqs),
           num_reqs_(num_reqs),
-          aligned_buf_(aligned_buf),
           dbg_(dbg) {}
 
     auto viaIfAsync(folly::Executor::KeepAlive<> executor) const {
       return folly::coro::co_viaIfAsync(
-          std::move(executor), Awaiter{reader_, file_, opts_, read_reqs_,
-                                       num_reqs_, aligned_buf_, dbg_});
+          std::move(executor),
+          Awaiter{reader_, file_, opts_, read_reqs_, num_reqs_, dbg_});
     }
 
    private:
@@ -127,7 +123,6 @@ class AsyncFileReader {
     const IOOptions& opts_;
     FSReadRequest* read_reqs_;
     size_t num_reqs_;
-    AlignedBuf* aligned_buf_;
     IODebugContext* dbg_;
   };
 

util/cleanable.cc

@@ -162,6 +162,10 @@ Cleanable* SharedCleanablePtr::get() {
   return ptr_;  // implicit upcast
 }
 
+const Cleanable* SharedCleanablePtr::get() const {
+  return ptr_;  // implicit upcast
+}
+
 void SharedCleanablePtr::RegisterCopyWith(Cleanable* target) {
   if (ptr_) {
     // "Virtual" copy of the pointer

util/io_dispatcher_imp.cc

@@ -14,6 +14,7 @@
 
 #include "util/io_dispatcher_imp.h"
 
+#include <algorithm>
 #include <deque>
 #include <memory>
 #include <unordered_map>
@@ -25,6 +26,7 @@
 #include "port/port.h"
 #include "rocksdb/file_system.h"
 #include "rocksdb/io_dispatcher.h"
+#include "rocksdb/io_status.h"
 #include "rocksdb/options.h"
 #include "rocksdb/status.h"
 #include "table/block_based/block_based_table_reader.h"
@@ -32,6 +34,7 @@
 #include "table/block_based/reader_common.h"
 #include "table/format.h"
 #include "test_util/sync_point.h"
+#include "util/aligned_buffer.h"
 #include "util/mutexlock.h"
 
 namespace ROCKSDB_NAMESPACE {
@@ -68,8 +71,13 @@ static bool BlockIndicesAreSortedByOffset(
 static Status CreateAndPinBlockFromBuffer(
     const std::shared_ptr<IOJob>& job, const BlockHandle& block,
     uint64_t buffer_start_offset, const Slice& buffer_data,
+    const SharedCleanablePtr& read_buffer_cleanup,
+    bool read_buffer_requires_cleanup,
     CachableEntry<Block>& pinned_block_entry) {
   auto* rep = job->table->get_rep();
+  const bool use_data_block_cache = ShouldUseDataBlockCacheForIterator(
+      rep->table_options, job->job_options.read_options,
+      rep->ioptions.allow_mmap_reads);
 
   // Get decompressor
   UnownedPtr<Decompressor> decompressor = rep->decompressor.get();
@@ -90,20 +98,112 @@ static Status CreateAndPinBlockFromBuffer(
   const auto block_size_with_trailer =
       BlockBasedTable::BlockSizeWithTrailer(block);
   const auto block_offset_in_buffer = block.offset() - buffer_start_offset;
+  const char* block_data = buffer_data.data() + block_offset_in_buffer;
 
-  CacheAllocationPtr data = AllocateBlock(
-      block_size_with_trailer, GetMemoryAllocator(rep->table_options));
-  memcpy(data.get(), buffer_data.data() + block_offset_in_buffer,
-         block_size_with_trailer);
-  BlockContents tmp_contents(std::move(data), block.size());
+  if (use_data_block_cache) {
+    CacheAllocationPtr data = AllocateBlock(
+        block_size_with_trailer, GetMemoryAllocator(rep->table_options));
+    memcpy(data.get(), block_data, block_size_with_trailer);
+    BlockContents tmp_contents(std::move(data), block.size());
 
 #ifndef NDEBUG
-  tmp_contents.has_trailer = rep->footer.GetBlockTrailerSize() > 0;
+    tmp_contents.has_trailer = rep->footer.GetBlockTrailerSize() > 0;
 #endif
 
-  return job->table->CreateAndPinBlockInCache<Block_kData>(
-      job->job_options.read_options, block, decompressor, &tmp_contents,
-      &pinned_block_entry.As<Block_kData>());
+    return job->table->CreateAndPinBlockInCache<Block_kData>(
+        job->job_options.read_options, block, decompressor, &tmp_contents,
+        &pinned_block_entry.As<Block_kData>());
+  }
+
+  BlockContents tmp_contents;
+  const CompressionType compression_type =
+      BlockBasedTable::GetBlockCompressionType(block_data, block.size());
+  ReadScopedBlockBufferProviderRef block_buffer_provider =
+      GetReadScopedBlockBufferProvider(job->job_options.read_options,
+                                       rep->ioptions.allow_mmap_reads);
+  Status s;
+  if (compression_type == kNoCompression) {
+    // Provider-backed uncompressed blocks should already be in the
+    // provider-owned read buffer allocated before I/O. Attach that cleanup to
+    // BlockContents so the block points at the read buffer without copying.
+#ifndef NDEBUG
+    SharedCleanablePtr test_read_buffer_cleanup = read_buffer_cleanup;
+    TEST_SYNC_POINT_CALLBACK("CreateAndPinBlockFromBuffer:ReadBufferCleanup",
+                             &test_read_buffer_cleanup);
+    const SharedCleanablePtr* effective_read_buffer_cleanup =
+        &test_read_buffer_cleanup;
+#else
+    const SharedCleanablePtr* effective_read_buffer_cleanup =
+        &read_buffer_cleanup;
+#endif
+    if (effective_read_buffer_cleanup->get() != nullptr) {
+      tmp_contents.data = Slice(block_data, block.size());
+      tmp_contents.cleanup = *effective_read_buffer_cleanup;
+      tmp_contents.backing_size = block_size_with_trailer;
+      tmp_contents.AssertSingleOwner();
+#ifndef NDEBUG
+      tmp_contents.has_trailer = rep->footer.GetBlockTrailerSize() > 0;
+#endif
+    } else if (read_buffer_requires_cleanup) {
+      s = Status::InvalidArgument(
+          "read-scoped block buffer provider requires read buffer cleanup");
+    } else if (block_buffer_provider.has_value()) {
+      s = CopyBufferToReadScopedBlockContents(
+          Slice(block_data, block_size_with_trailer), block.size(),
+          block_buffer_provider->get(), &tmp_contents);
+    } else {
+      s = CopyBufferToHeapBlockContents(
+          Slice(block_data, block_size_with_trailer), block.size(),
+          GetMemoryAllocator(rep->table_options), &tmp_contents);
+    }
+  } else {
+    // Compressed blocks cannot view the read buffer as final block contents.
+    // When a provider is configured, decompression allocates provider-backed
+    // output and writes the uncompressed block directly into it.
+    s = DecompressSerializedBlock(block_data, block.size(), compression_type,
+                                  *decompressor, &tmp_contents, rep->ioptions,
+                                  GetMemoryAllocator(rep->table_options),
+                                  block_buffer_provider);
+  }
+  if (!s.ok()) {
+    return s;
+  }
+
+  std::unique_ptr<Block_kData> block_holder;
+  rep->create_context.Create(&block_holder, std::move(tmp_contents));
+  pinned_block_entry.As<Block_kData>().SetOwnedValue(std::move(block_holder));
+  return Status::OK();
+}
+
+struct ReadScopedIOConfig {
+  bool use_data_block_cache = true;
+  ReadScopedBlockBufferProviderRef block_buffer_provider;
+  // True only when provider-backed file-read scratch is required because the
+  // block is known to be uncompressed. If this is true, the read buffer cleanup
+  // must be attached directly to BlockContents; otherwise maybe-compressed
+  // reads may use ordinary scratch and copy/decompress into provider-backed
+  // final contents after the compression type is known.
+  bool read_buffer_requires_cleanup = false;
+  bool use_read_scoped_direct_io = false;
+  bool use_read_scoped_scratch = false;
+};
+
+static ReadScopedIOConfig GetReadScopedIOConfig(
+    const BlockBasedTableOptions& table_options, const JobOptions& job_options,
+    bool allow_mmap_reads, bool data_blocks_maybe_compressed, bool direct_io) {
+  ReadScopedIOConfig config;
+  config.use_data_block_cache = ShouldUseDataBlockCacheForIterator(
+      table_options, job_options.read_options, allow_mmap_reads);
+  config.block_buffer_provider = GetReadScopedBlockBufferProvider(
+      job_options.read_options, allow_mmap_reads);
+
+  const bool use_read_scoped_buffer =
+      !config.use_data_block_cache &&
+      config.block_buffer_provider.has_value() && !data_blocks_maybe_compressed;
+  config.read_buffer_requires_cleanup = use_read_scoped_buffer;
+  config.use_read_scoped_direct_io = direct_io && use_read_scoped_buffer;
+  config.use_read_scoped_scratch = !direct_io && use_read_scoped_buffer;
+  return config;
 }
 
 // State for async IO operations (implementation detail)
@@ -116,11 +216,16 @@ struct AsyncIOState {
   AsyncIOState(AsyncIOState&&) = default;
   AsyncIOState& operator=(AsyncIOState&&) = default;
 
+  ReadScopedBlockBufferProvider::Lease read_scoped_buf_lease;
   std::unique_ptr<char[]> buf;
   AlignedBuf aligned_buf;
+  AlignedBuffer direct_io_buffer;
   void* io_handle = nullptr;
   IOHandleDeleter del_fn;
   uint64_t offset;
+  // Captures ReadScopedIOConfig::read_buffer_requires_cleanup until async I/O
+  // completion processing can attach the read buffer cleanup to BlockContents.
+  bool read_buffer_requires_cleanup = false;
   std::vector<size_t> block_indices;
   std::vector<BlockHandle> blocks;
   FSReadRequest read_req;
@@ -130,15 +235,13 @@ struct AsyncIOState {
 // Must call AbortIO before deleting handles to avoid use-after-free when
 // io_uring completions arrive for deleted handles.
 ReadSet::~ReadSet() {
-  // Release memory for any blocks still pinned
-  // Note: block_sizes_[i] is only set for async IO reads where memory
-  // limiting applies. For sync reads, block_sizes_ remains 0, so this
-  // loop is effectively a no-op for sync reads.
-  if (auto dispatcher_data = dispatcher_data_.lock()) {
-    for (size_t i = 0; i < block_sizes_.size(); ++i) {
-      if (block_sizes_[i] > 0 && pinned_blocks_[i].GetValue()) {
-        dispatcher_data->ReleaseMemory(block_sizes_[i]);
-      }
+  // Release memory for any blocks still owned by this ReadSet. Pending queued
+  // blocks have a non-zero block_sizes_ entry, but no memory was acquired for
+  // them yet, so only pinned or async-dispatched blocks should release memory.
+  for (size_t i = 0; i < block_sizes_.size(); ++i) {
+    if (pinned_blocks_[i].GetValue() ||
+        async_io_map_.find(i) != async_io_map_.end()) {
+      ReleasePrefetchMemory(i);
     }
   }
 
@@ -168,11 +271,7 @@ ReadSet::~ReadSet() {
 
   // Now safe to delete the handles
   for (auto& pair : async_io_map_) {
-    auto& async_state = pair.second;
-    if (async_state->io_handle != nullptr && async_state->del_fn != nullptr) {
-      async_state->del_fn(async_state->io_handle);
-      async_state->io_handle = nullptr;
-    }
+    DeleteAsyncIOHandle(pair.second);
   }
 }
 
@@ -190,12 +289,7 @@ Status ReadSet::ReadIndex(size_t block_index, CachableEntry<Block>* out) {
     // Release memory accounting for prefetched blocks. After moving the value
     // out, ReleaseBlock() and the destructor check pinned_blocks_.GetValue()
     // which will be null, so they won't release memory again.
-    if (block_index < block_sizes_.size() && block_sizes_[block_index] > 0) {
-      if (auto dispatcher_data = dispatcher_data_.lock()) {
-        dispatcher_data->ReleaseMemory(block_sizes_[block_index]);
-      }
-      block_sizes_[block_index] = 0;
-    }
+    ReleasePrefetchMemory(block_index);
     // Note: Statistics for this block were already counted during SubmitJob
     // (either as cache hit or sync read)
     return Status::OK();
@@ -219,13 +313,7 @@ Status ReadSet::ReadIndex(size_t block_index, CachableEntry<Block>* out) {
       if (pinned_blocks_[block_index].GetValue()) {
         *out = std::move(pinned_blocks_[block_index]);
         // Release memory accounting (same as case 1 above)
-        if (block_index < block_sizes_.size() &&
-            block_sizes_[block_index] > 0) {
-          if (auto dispatcher_data = dispatcher_data_.lock()) {
-            dispatcher_data->ReleaseMemory(block_sizes_[block_index]);
-          }
-          block_sizes_[block_index] = 0;
-        }
+        ReleasePrefetchMemory(block_index);
         return Status::OK();
       }
 
@@ -291,22 +379,68 @@ void ReadSet::ReleaseBlock(size_t block_index) {
   // Remove from pending if applicable
   RemoveFromPending(block_index);
 
-  // Release memory BEFORE unpinning
-  // Note: block_sizes_[idx] is only set for async IO reads where memory
-  // limiting applies. For sync reads, block_sizes_ remains 0, so this
-  // check implicitly skips ReleaseMemory for sync reads.
-  if (pinned_blocks_[block_index].GetValue() &&
-      block_index < block_sizes_.size() && block_sizes_[block_index] > 0) {
-    if (auto dispatcher_data = dispatcher_data_.lock()) {
-      dispatcher_data->ReleaseMemory(block_sizes_[block_index]);
-    }
-    block_sizes_[block_index] = 0;  // Prevent double-release
+  // Release memory for a materialized prefetched block before unpinning. Queued
+  // blocks have not acquired memory yet, and pending async blocks release their
+  // memory budget in ReleaseAsyncIOForBlock().
+  if (pinned_blocks_[block_index].GetValue()) {
+    ReleasePrefetchMemory(block_index);
   }
 
   // Unpin the block from cache
   pinned_blocks_[block_index].Reset();
-  // Clean up any pending async IO for this block
-  async_io_map_.erase(block_index);
+  ReleaseAsyncIOForBlock(block_index);
+}
+
+void ReadSet::ReleasePrefetchMemory(size_t block_index) {
+  if (block_index >= block_sizes_.size() || block_sizes_[block_index] == 0) {
+    return;
+  }
+
+  if (auto dispatcher_data = dispatcher_data_.lock()) {
+    dispatcher_data->ReleaseMemory(block_sizes_[block_index]);
+  }
+  block_sizes_[block_index] = 0;
+}
+
+void ReadSet::ReleaseAsyncIOForBlock(size_t block_index) {
+  auto map_iter = async_io_map_.find(block_index);
+  if (map_iter == async_io_map_.end()) {
+    return;
+  }
+
+  std::shared_ptr<AsyncIOState> async_state = map_iter->second;
+  async_io_map_.erase(map_iter);
+  ReleasePrefetchMemory(block_index);
+
+  auto block_iter = std::find(async_state->block_indices.begin(),
+                              async_state->block_indices.end(), block_index);
+  if (block_iter != async_state->block_indices.end()) {
+    const size_t state_index =
+        static_cast<size_t>(block_iter - async_state->block_indices.begin());
+    async_state->block_indices.erase(block_iter);
+    async_state->blocks.erase(async_state->blocks.begin() + state_index);
+  }
+
+  if (!async_state->block_indices.empty()) {
+    return;
+  }
+
+  if (async_state->io_handle != nullptr && fs_ != nullptr) {
+    std::vector<void*> io_handles = {async_state->io_handle};
+    TEST_SYNC_POINT_CALLBACK("ReadSet::ReleaseAsyncIOForBlock:AbortIO",
+                             nullptr);
+    IOStatus s = fs_->AbortIO(io_handles);
+    s.PermitUncheckedError();
+  }
+  DeleteAsyncIOHandle(async_state);
+}
+
+void ReadSet::DeleteAsyncIOHandle(
+    const std::shared_ptr<AsyncIOState>& async_state) {
+  if (async_state->io_handle != nullptr && async_state->del_fn != nullptr) {
+    async_state->del_fn(async_state->io_handle);
+    async_state->io_handle = nullptr;
+  }
 }
 
 bool ReadSet::IsBlockAvailable(size_t block_index) const {
@@ -339,25 +473,27 @@ Status ReadSet::PollAndProcessAsyncIO(
   // Use the result slice from the callback which has been correctly set
   // with any necessary alignment adjustments for direct IO
   const Slice& buffer_data = async_state->read_req.result;
+  SharedCleanablePtr read_buffer_cleanup;
+  if (async_state->read_scoped_buf_lease.cleanup.get() != nullptr) {
+    read_buffer_cleanup = std::move(async_state->read_scoped_buf_lease.cleanup);
+  }
 
   // Process all blocks in this async request
   for (size_t i = 0; i < async_state->block_indices.size(); ++i) {
     const size_t idx = async_state->block_indices[i];
     const auto& block_handle = async_state->blocks[i];
 
-    Status s =
-        CreateAndPinBlockFromBuffer(job_, block_handle, async_state->offset,
-                                    buffer_data, pinned_blocks_[idx]);
+    Status s = CreateAndPinBlockFromBuffer(
+        job_, block_handle, async_state->offset, buffer_data,
+        read_buffer_cleanup, async_state->read_buffer_requires_cleanup,
+        pinned_blocks_[idx]);
     if (!s.ok()) {
       return s;
     }
   }
 
   // Clean up IO handle
-  if (async_state->io_handle != nullptr && async_state->del_fn != nullptr) {
-    async_state->del_fn(async_state->io_handle);
-    async_state->io_handle = nullptr;
-  }
+  DeleteAsyncIOHandle(async_state);
 
   // Remove from map - all blocks in this request have been processed
   // Store indices in a temporary vector to avoid iterator invalidation
@@ -375,6 +511,9 @@ Status ReadSet::PollAndProcessAsyncIO(
 Status ReadSet::SyncRead(size_t block_index) {
   const auto& block_handle = job_->block_handles[block_index];
   auto* rep = job_->table->get_rep();
+  const bool use_data_block_cache = ShouldUseDataBlockCacheForIterator(
+      rep->table_options, job_->job_options.read_options,
+      rep->ioptions.allow_mmap_reads);
 
   // Get dictionary-aware decompressor if available
   UnownedPtr<Decompressor> decompressor = rep->decompressor.get();
@@ -395,8 +534,8 @@ Status ReadSet::SyncRead(size_t block_index) {
       /*prefetch_buffer=*/nullptr, job_->job_options.read_options, block_handle,
       decompressor, &pinned_blocks_[block_index].As<Block_kData>(),
       /*get_context=*/nullptr, /*lookup_context=*/nullptr,
-      /*for_compaction=*/false, /*use_cache=*/true,
-      /*async_read=*/false, /*use_block_cache_for_lookup=*/true);
+      /*for_compaction=*/false, use_data_block_cache,
+      /*async_read=*/false, use_data_block_cache);
 }
 
 // A pre-coalesced group of blocks for prefetching
@@ -731,22 +870,29 @@ Status IODispatcherImpl::Impl::SubmitJob(const std::shared_ptr<IOJob>& job,
   // sorted.
   std::vector<size_t> block_indices_to_read;
   block_indices_to_read.reserve(job->block_handles.size());
+  const bool use_data_block_cache = ShouldUseDataBlockCacheForIterator(
+      job->table->get_rep()->table_options, job->job_options.read_options,
+      job->table->get_rep()->ioptions.allow_mmap_reads);
 
-  for (size_t i : rs->sorted_block_indices_) {
-    const auto& data_block_handle = job->block_handles[i];
+  if (!use_data_block_cache) {
+    block_indices_to_read = rs->sorted_block_indices_;
+  } else {
+    for (size_t i : rs->sorted_block_indices_) {
+      const auto& data_block_handle = job->block_handles[i];
 
-    // Lookup and pin block in cache
-    Status s = job->table->LookupAndPinBlocksInCache<Block_kData>(
-        job->job_options.read_options, data_block_handle,
-        &(rs->pinned_blocks_)[i].As<Block_kData>());
+      // Lookup and pin block in cache
+      Status s = job->table->LookupAndPinBlocksInCache<Block_kData>(
+          job->job_options.read_options, data_block_handle,
+          &(rs->pinned_blocks_)[i].As<Block_kData>());
 
-    if (!s.ok()) {
-      continue;
-    }
+      if (!s.ok()) {
+        continue;
+      }
 
-    if (!(rs->pinned_blocks_)[i].GetValue()) {
-      // Block not in cache - needs to be read from disk
-      block_indices_to_read.emplace_back(i);
+      if (!(rs->pinned_blocks_)[i].GetValue()) {
+        // Block not in cache - needs to be read from disk
+        block_indices_to_read.emplace_back(i);
+      }
     }
   }
 
@@ -966,12 +1112,17 @@ std::vector<size_t> IODispatcherImpl::Impl::ExecuteAsyncIO(
   }
 
   const bool direct_io = rep->file->use_direct_io();
+  const ReadScopedIOConfig read_scoped_io = GetReadScopedIOConfig(
+      rep->table_options, job->job_options, rep->ioptions.allow_mmap_reads,
+      rep->decompressor != nullptr, direct_io);
 
   // Submit async read requests and store them in the ReadSet
   for (size_t i = 0; i < read_reqs.size(); ++i) {
     auto async_state = std::make_shared<AsyncIOState>();
 
     async_state->offset = read_reqs[i].offset;
+    async_state->read_buffer_requires_cleanup =
+        read_scoped_io.read_buffer_requires_cleanup;
     async_state->block_indices = coalesced_block_indices[i];
     async_state->read_req = std::move(read_reqs[i]);
 
@@ -979,8 +1130,38 @@ std::vector<size_t> IODispatcherImpl::Impl::ExecuteAsyncIO(
       async_state->blocks.emplace_back(job->block_handles[idx]);
     }
 
+    AlignedBuffer::Allocator direct_io_allocator;
+    AlignedBufferAllocationContext direct_io_context{
+        &async_state->direct_io_buffer};
+    AlignedBufferAllocationContext* direct_io_context_arg = nullptr;
+    if (read_scoped_io.use_read_scoped_direct_io) {
+      assert(read_scoped_io.block_buffer_provider.has_value());
+      // This allocator borrows the AsyncIOState lease member and is invoked
+      // synchronously by RandomAccessFileReader::ReadAsync before it returns.
+      // The resulting AlignedBuffer owner keeps the read-scoped cleanup alive
+      // until the async read is processed or abandoned. Uncompressed blocks
+      // later attach this cleanup directly to BlockContents.
+      direct_io_allocator = MakeReadScopedAlignedBufferAllocator(
+          read_scoped_io.block_buffer_provider,
+          &async_state->read_scoped_buf_lease);
+      direct_io_context.allocator = &direct_io_allocator;
+      direct_io_context_arg = &direct_io_context;
+    }
+
     if (direct_io) {
       async_state->read_req.scratch = nullptr;
+    } else if (read_scoped_io.use_read_scoped_scratch) {
+      assert(read_scoped_io.block_buffer_provider.has_value());
+      // Non-direct I/O writes into provider-backed scratch. For uncompressed
+      // blocks this scratch becomes the final BlockContents backing.
+      s = AllocateReadScopedBlockBuffer(
+          read_scoped_io.block_buffer_provider->get(),
+          async_state->read_req.len, 1, &async_state->read_scoped_buf_lease);
+      if (!s.ok()) {
+        *out_status = s;
+        return fallback_block_indices;
+      }
+      async_state->read_req.scratch = async_state->read_scoped_buf_lease.data;
     } else {
       async_state->buf.reset(new char[async_state->read_req.len]);
       async_state->read_req.scratch = async_state->buf.get();
@@ -995,10 +1176,13 @@ std::vector<size_t> IODispatcherImpl::Impl::ExecuteAsyncIO(
       state->read_req.status = req.status;
     };
 
-    s = rep->file->ReadAsync(async_state->read_req, io_opts, cb,
-                             async_state.get(), &async_state->io_handle,
-                             &async_state->del_fn,
-                             direct_io ? &async_state->aligned_buf : nullptr);
+    s = rep->file->ReadAsync(
+        async_state->read_req, io_opts, cb, async_state.get(),
+        &async_state->io_handle, &async_state->del_fn,
+        read_scoped_io.use_read_scoped_direct_io
+            ? nullptr
+            : (direct_io ? &async_state->aligned_buf : nullptr),
+        /*dbg=*/nullptr, direct_io_context_arg);
 
     if (s.IsNotSupported()) {
       // Async IO may be compiled in but unavailable at runtime. Fall back to
@@ -1036,24 +1220,76 @@ Status IODispatcherImpl::Impl::ExecuteSyncIO(
     const std::shared_ptr<IOJob>& job, const std::shared_ptr<ReadSet>& read_set,
     std::vector<FSReadRequest>& read_reqs,
     const std::vector<std::vector<size_t>>& coalesced_block_indices) {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+  // Some error exits intentionally ignore FSReadRequest::status values: setup
+  // failures happen before MultiRead populates them, top-level MultiRead
+  // errors make per-request statuses irrelevant, and production short-circuits
+  // on the first per-request error.
+  auto permit_unchecked_read_req_statuses = [&read_reqs]() {
+    for (const FSReadRequest& read_req : read_reqs) {
+      read_req.status.PermitUncheckedError();
+    }
+  };
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
+
   // Get file and IO options
   auto* rep = job->table->get_rep();
   IOOptions io_opts;
   if (Status s =
           rep->file->PrepareIOOptions(job->job_options.read_options, io_opts);
       !s.ok()) {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+    permit_unchecked_read_req_statuses();
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
     return s;
   }
 
   const bool direct_io = rep->file->use_direct_io();
+  const ReadScopedIOConfig read_scoped_io = GetReadScopedIOConfig(
+      rep->table_options, job->job_options, rep->ioptions.allow_mmap_reads,
+      rep->decompressor != nullptr, direct_io);
 
   // Setup scratch buffers for MultiRead
   std::unique_ptr<char[]> buf;
+  std::vector<ReadScopedBlockBufferProvider::Lease> read_scoped_leases;
+  std::vector<SharedCleanablePtr> read_req_cleanups;
+
+  ReadScopedBlockBufferProvider::Lease read_scoped_direct_io_lease;
+  AlignedBuffer direct_io_buffer;
+  AlignedBuffer::Allocator direct_io_allocator;
+  AlignedBufferAllocationContext direct_io_context{&direct_io_buffer};
+  if (read_scoped_io.use_read_scoped_direct_io) {
+    assert(read_scoped_io.block_buffer_provider.has_value());
+    // This allocator borrows stack-local lease state from the synchronous
+    // ExecuteSyncIO call. RandomAccessFileReader::MultiRead asks it to allocate
+    // before returning, and uncompressed blocks later attach the lease cleanup
+    // directly to BlockContents.
+    direct_io_allocator = MakeReadScopedAlignedBufferAllocator(
+        read_scoped_io.block_buffer_provider, &read_scoped_direct_io_lease);
+    direct_io_context.allocator = &direct_io_allocator;
+  }
 
   if (direct_io) {
     for (auto& read_req : read_reqs) {
       read_req.scratch = nullptr;
     }
+  } else if (read_scoped_io.use_read_scoped_scratch) {
+    assert(read_scoped_io.block_buffer_provider.has_value());
+    // Non-direct I/O writes into provider-backed scratch. For uncompressed
+    // blocks this scratch becomes the final BlockContents backing.
+    read_scoped_leases.resize(read_reqs.size());
+    for (size_t i = 0; i < read_reqs.size(); ++i) {
+      if (Status s = AllocateReadScopedBlockBuffer(
+              read_scoped_io.block_buffer_provider->get(), read_reqs[i].len, 1,
+              &read_scoped_leases[i]);
+          !s.ok()) {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+        permit_unchecked_read_req_statuses();
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
+        return s;
+      }
+      read_reqs[i].scratch = read_scoped_leases[i].data;
+    }
   } else {
     // Allocate a single contiguous buffer for all requests
     size_t total_len = 0;
@@ -1069,20 +1305,35 @@ Status IODispatcherImpl::Impl::ExecuteSyncIO(
   }
 
   // Execute MultiRead
-  AlignedBuf aligned_buf;
   if (Status s =
           rep->file->MultiRead(io_opts, read_reqs.data(), read_reqs.size(),
-                               direct_io ? &aligned_buf : nullptr);
+                               &direct_io_context, /*dbg=*/nullptr);
       !s.ok()) {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+    permit_unchecked_read_req_statuses();
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
     return s;
   }
 
   for (const auto& rq : read_reqs) {
     if (!rq.status.ok()) {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+      permit_unchecked_read_req_statuses();
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
       return rq.status;
     }
   }
 
+  if (read_scoped_io.use_read_scoped_direct_io) {
+    read_req_cleanups.assign(read_reqs.size(),
+                             read_scoped_direct_io_lease.cleanup);
+  } else if (read_scoped_io.use_read_scoped_scratch) {
+    read_req_cleanups.resize(read_reqs.size());
+    for (size_t i = 0; i < read_scoped_leases.size(); ++i) {
+      read_req_cleanups[i] = std::move(read_scoped_leases[i].cleanup);
+    }
+  }
+
   // Process all blocks from the MultiRead results
   for (size_t i = 0; i < coalesced_block_indices.size(); ++i) {
     const auto& read_req = read_reqs[i];
@@ -1091,6 +1342,9 @@ Status IODispatcherImpl::Impl::ExecuteSyncIO(
 
       Status create_status = CreateAndPinBlockFromBuffer(
           job, block_handle, read_req.offset, read_req.result,
+          i < read_req_cleanups.size() ? read_req_cleanups[i]
+                                       : SharedCleanablePtr(),
+          read_scoped_io.read_buffer_requires_cleanup,
           read_set->pinned_blocks_[block_idx]);
       if (!create_status.ok()) {
         return create_status;

utilities/blob_db/blob_db_impl.cc

@@ -1390,7 +1390,7 @@ Status BlobDBImpl::GetRawBlobFromFile(const Slice& key, uint64_t file_number,
 
   // Allocate the buffer. This is safe in C++11
   std::string buf;
-  AlignedBuf aligned_buf;
+  AlignedBuffer direct_io_buffer;
 
   // A partial blob record contain checksum, key and value.
   Slice blob_record;
@@ -1399,14 +1399,15 @@ Status BlobDBImpl::GetRawBlobFromFile(const Slice& key, uint64_t file_number,
     StopWatch read_sw(clock_, statistics_, BLOB_DB_BLOB_FILE_READ_MICROS);
     // TODO: rate limit old blob DB file reads.
     if (reader->use_direct_io()) {
+      AlignedBufferAllocationContext direct_io_context{&direct_io_buffer};
       s = reader->Read(IOOptions(), record_offset,
                        static_cast<size_t>(record_size), &blob_record, nullptr,
-                       &aligned_buf);
+                       &direct_io_context);
     } else {
       buf.reserve(static_cast<size_t>(record_size));
       s = reader->Read(IOOptions(), record_offset,
                        static_cast<size_t>(record_size), &blob_record,
-                       buf.data(), nullptr);
+                       buf.data());
     }
     RecordTick(statistics_, BLOB_DB_BLOB_FILE_BYTES_READ, blob_record.size());
   }

utilities/blob_db/blob_file.cc

@@ -104,16 +104,17 @@ Status BlobFile::ReadFooter(BlobLogFooter* bf) {
 
   Slice result;
   std::string buf;
-  AlignedBuf aligned_buf;
+  AlignedBuffer direct_io_buffer;
   Status s;
   // TODO: rate limit reading footers from blob files.
   if (ra_file_reader_->use_direct_io()) {
+    AlignedBufferAllocationContext direct_io_context{&direct_io_buffer};
     s = ra_file_reader_->Read(IOOptions(), footer_offset, BlobLogFooter::kSize,
-                              &result, nullptr, &aligned_buf);
+                              &result, nullptr, &direct_io_context);
   } else {
     buf.reserve(BlobLogFooter::kSize + 10);
     s = ra_file_reader_->Read(IOOptions(), footer_offset, BlobLogFooter::kSize,
-                              &result, buf.data(), nullptr);
+                              &result, buf.data());
   }
   if (!s.ok()) {
     return s;
@@ -228,16 +229,17 @@ Status BlobFile::ReadMetadata(const std::shared_ptr<FileSystem>& fs,
 
   // Read file header.
   std::string header_buf;
-  AlignedBuf aligned_buf;
+  AlignedBuffer direct_io_buffer;
   Slice header_slice;
   // TODO: rate limit reading headers from blob files.
   if (file_reader->use_direct_io()) {
+    AlignedBufferAllocationContext direct_io_context{&direct_io_buffer};
     s = file_reader->Read(IOOptions(), 0, BlobLogHeader::kSize, &header_slice,
-                          nullptr, &aligned_buf);
+                          nullptr, &direct_io_context);
   } else {
     header_buf.reserve(BlobLogHeader::kSize);
     s = file_reader->Read(IOOptions(), 0, BlobLogHeader::kSize, &header_slice,
-                          header_buf.data(), nullptr);
+                          header_buf.data());
   }
   if (!s.ok()) {
     ROCKS_LOG_ERROR(
@@ -271,14 +273,15 @@ Status BlobFile::ReadMetadata(const std::shared_ptr<FileSystem>& fs,
   Slice footer_slice;
   // TODO: rate limit reading footers from blob files.
   if (file_reader->use_direct_io()) {
+    AlignedBufferAllocationContext direct_io_context{&direct_io_buffer};
     s = file_reader->Read(IOOptions(), file_size - BlobLogFooter::kSize,
                           BlobLogFooter::kSize, &footer_slice, nullptr,
-                          &aligned_buf);
+                          &direct_io_context);
   } else {
     footer_buf.reserve(BlobLogFooter::kSize);
     s = file_reader->Read(IOOptions(), file_size - BlobLogFooter::kSize,
                           BlobLogFooter::kSize, &footer_slice,
-                          footer_buf.data(), nullptr);
+                          footer_buf.data());
   }
   if (!s.ok()) {
     ROCKS_LOG_ERROR(