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

.github/actions/windows-build-steps/action.yml

@@ -3,7 +3,7 @@ inputs:
   suite-run:
     description: Comma-separated list of test suites to run (empty to skip C++ tests)
     required: false
-    default: arena_test,db_basic_test,db_test,db_test2,db_merge_operand_test,bloom_test,c_test,coding_test,crc32c_test,dynamic_bloom_test,env_basic_test,env_test,hash_test,random_test
+    default: arena_test,db_basic_test,db_test,db_etc2_test,db_merge_operand_test,bloom_test,c_test,coding_test,crc32c_test,dynamic_bloom_test,env_basic_test,env_test,hash_test,random_test
   run-java:
     description: Whether to run Java tests
     required: false

.github/workflows/pr-jobs.yml

@@ -537,7 +537,7 @@ jobs:
           suite_run: db_test
           run_java: "false"
         - test_shard: other
-          suite_run: arena_test,db_basic_test,db_test2,db_merge_operand_test,bloom_test,c_test,coding_test,crc32c_test,dynamic_bloom_test,env_basic_test,env_test,hash_test,random_test
+          suite_run: arena_test,db_basic_test,db_etc2_test,db_merge_operand_test,bloom_test,c_test,coding_test,crc32c_test,dynamic_bloom_test,env_basic_test,env_test,hash_test,random_test
           run_java: "false"
         - test_shard: java
           suite_run: ""

.gitignore

@@ -52,7 +52,7 @@ GRTAGS
 GTAGS
 rocksdb_dump
 rocksdb_undump
-db_test2
+db_etc2_test
 trace_analyzer
 block_cache_trace_analyzer
 io_tracer_parser

BUCK

@@ -4868,6 +4868,12 @@ cpp_unittest_wrapper(name="db_encryption_test",
             extra_compiler_flags=[])
 
 
+cpp_unittest_wrapper(name="db_etc2_test",
+            srcs=["db/db_etc2_test.cc"],
+            deps=[":rocksdb_test_lib"],
+            extra_compiler_flags=[])
+
+
 cpp_unittest_wrapper(name="db_etc3_test",
             srcs=["db/db_etc3_test.cc"],
             deps=[":rocksdb_test_lib"],
@@ -5024,12 +5030,6 @@ cpp_unittest_wrapper(name="db_test",
             extra_compiler_flags=[])
 
 
-cpp_unittest_wrapper(name="db_test2",
-            srcs=["db/db_test2.cc"],
-            deps=[":rocksdb_test_lib"],
-            extra_compiler_flags=[])
-
-
 cpp_unittest_wrapper(name="db_universal_compaction_test",
             srcs=["db/db_universal_compaction_test.cc"],
             deps=[":rocksdb_test_lib"],

CLAUDE.md

@@ -209,12 +209,11 @@ The following patterns emerged as frequent sources of review feedback:
 ## Important tips
 
 ### Build system
-* There are 3 build system. Make, CMake, BUCK(meta internal).
+* There are 3 build system. Make for git clones, BUCK (meta internal) for hg
+  clones, and CMake for some special cases.
 * When a new .cc file is added, update Makefile, CMakeLists.txt, src.mk, BUCK.
 * Don't manually edit BUCK file, after updating src.mk, run
     /usr/local/bin/python3 buckifier/buckify_rocksdb.py to update it
-* Use make to build and run the test. CMake and BUCK are not used locally.
-* Use `make dbg` command to build all of the unit test in debug mode.
 * For -j in make command, use the number of CPU cores to decide it.
 
 ### When to run `make clean` (avoid mixing build modes)

CMakeLists.txt

@@ -1491,6 +1491,7 @@ if(WITH_TESTS)
         db/db_clip_test.cc
         db/db_dynamic_level_test.cc
         db/db_encryption_test.cc
+        db/db_etc2_test.cc
         db/db_etc3_test.cc
         db/db_flush_test.cc
         db/db_inplace_update_test.cc
@@ -1514,7 +1515,6 @@ if(WITH_TESTS)
         db/db_table_properties_test.cc
         db/db_tailing_iter_test.cc
         db/db_test.cc
-        db/db_test2.cc
         db/db_logical_block_size_cache_test.cc
         db/db_universal_compaction_test.cc
         db/db_wal_test.cc

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.

Makefile

@@ -967,8 +967,6 @@ gen_parallel_tests:
 #      total time across many shards (need early queueing for fan-out).
 # Each alternative is wrapped in `^.*NAME.*$$` so the *whole input line* is
 # captured into $$1; then `s,(...),100 $$1,` prepends "100 " to the line.
-# Use a trailing dash on prefix-of-other-name binaries (e.g. `db_test-` to
-# avoid matching `db_test2-...`).
 #
 # Tiers below are based on observed timings (see suggest-slow-tests).
 # Tier 1: max single-shard time >= 30s (critical-path bottlenecks).
@@ -976,7 +974,7 @@ gen_parallel_tests:
 # Tier 3: huge total time across many tiny shards; front-loading them keeps
 #         the tail of the run busy while big shards finish.
 slow_test_regexp = \
-	^.*point_lock_manager_stress_test.*$$|^.*db_test-.*$$|^.*external_sst_file_test.*$$|^.*compaction_service_test.*$$|^.*corruption_test.*$$|^.*comparator_db_test.*$$|^.*external_sst_file_basic_test.*$$|^.*rate_limiter_test.*$$|^.*db_compaction_test.*$$|^.*write_prepared_transaction_test.*$$|^.*db_merge_operator_test.*$$|\
+	^.*point_lock_manager_stress_test.*$$|^.*db_test.*$$|^.*external_sst_file_test.*$$|^.*compaction_service_test.*$$|^.*corruption_test.*$$|^.*comparator_db_test.*$$|^.*external_sst_file_basic_test.*$$|^.*rate_limiter_test.*$$|^.*db_compaction_test.*$$|^.*write_prepared_transaction_test.*$$|^.*db_merge_operator_test.*$$|\
 	^.*db_dynamic_level_test.*$$|^.*db_bloom_filter_test.*$$|^.*error_handler_fs_test.*$$|^.*merge_helper_test.*$$|^.*transaction_test.*$$|^.*db_kv_checksum_test.*$$|^.*inlineskiplist_test.*$$|\
 	^.*db_with_timestamp_basic_test.*$$|^.*table_test.*$$|^.*db_wal_test.*$$|^.*block_based_table_reader_test.*$$|^.*block_test.*$$
 prioritize_long_running_tests =						\
@@ -1586,7 +1584,7 @@ db_open_with_config_test: $(OBJ_DIR)/db/db_open_with_config_test.o $(TEST_LIBRAR
 db_test: $(OBJ_DIR)/db/db_test.o $(TEST_LIBRARY) $(LIBRARY)
 	$(AM_LINK)
 
-db_test2: $(OBJ_DIR)/db/db_test2.o $(TEST_LIBRARY) $(LIBRARY)
+db_etc2_test: $(OBJ_DIR)/db/db_etc2_test.o $(TEST_LIBRARY) $(LIBRARY)
 	$(AM_LINK)
 
 db_etc3_test: $(OBJ_DIR)/db/db_etc3_test.o $(TEST_LIBRARY) $(LIBRARY)

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/compaction/compaction_service_job.cc

@@ -651,6 +651,10 @@ static std::unordered_map<std::string, OptionTypeInfo>
         {"cpu_micros",
          {offsetof(struct CompactionJobStats, cpu_micros), OptionType::kUInt64T,
           OptionVerificationType::kNormal, OptionTypeFlags::kNone}},
+        {"has_accurate_num_input_records",
+         {offsetof(struct CompactionJobStats, has_accurate_num_input_records),
+          OptionType::kBoolean, OptionVerificationType::kNormal,
+          OptionTypeFlags::kNone}},
         {"num_input_records",
          {offsetof(struct CompactionJobStats, num_input_records),
           OptionType::kUInt64T, OptionVerificationType::kNormal,

db/compaction/compaction_service_test.cc

@@ -943,6 +943,37 @@ TEST_F(CompactionServiceTest, VerifyInputRecordCount) {
   SyncPoint::GetInstance()->ClearAllCallBacks();
 }
 
+TEST_F(CompactionServiceTest, InaccurateInputRecordCount) {
+  Options options = CurrentOptions();
+  options.disable_auto_compactions = true;
+  ReopenWithCompactionService(&options);
+  GenerateTestData();
+
+  auto my_cs = GetCompactionService();
+
+  std::string start_str = Key(15);
+  std::string end_str = Key(45);
+  Slice start(start_str);
+  Slice end(end_str);
+  uint64_t comp_num = my_cs->GetCompactionNum();
+
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack(
+      "CompactionServiceCompactionJob::Run:0", [&](void* arg) {
+        CompactionServiceResult* compaction_result =
+            *(static_cast<CompactionServiceResult**>(arg));
+        ASSERT_TRUE(compaction_result != nullptr);
+        compaction_result->stats.has_accurate_num_input_records = false;
+        compaction_result->stats.num_input_records = 0;
+      });
+  SyncPoint::GetInstance()->EnableProcessing();
+
+  ASSERT_OK(db_->CompactRange(CompactRangeOptions(), &start, &end));
+  ASSERT_GE(my_cs->GetCompactionNum(), comp_num + 1);
+
+  SyncPoint::GetInstance()->DisableProcessing();
+  SyncPoint::GetInstance()->ClearAllCallBacks();
+}
+
 TEST_F(CompactionServiceTest, EmptyResult) {
   Options options = CurrentOptions();
   options.disable_auto_compactions = true;

db/db_bloom_filter_test.cc

@@ -2022,7 +2022,11 @@ TEST_F(DBBloomFilterTest, MutableFilterPolicy) {
   double expected_bpk = 10.0;
   // Other configs to try
   std::vector<std::pair<std::string, double>> configs = {
-      {"ribbonfilter:10:-1", 7.0}, {"bloomfilter:5", 5.0}, {"nullptr", 0.0}};
+      {"ribbonfilter:10:-1", 7.0},
+      {"bloomfilter:5", 5.0},
+      {"nullptr", 0.0},
+      // As serialized in OPTIONS file
+      {"{id=ribbonfilter:12:-1;bloom_before_level=-1;}", 8.4}};
 
   table_options.cache_index_and_filter_blocks = true;
   options.table_factory.reset(NewBlockBasedTableFactory(table_options));

db/db_compaction_abort_test.cc

@@ -427,6 +427,38 @@ TEST_F(DBCompactionAbortTest, AbortAutomaticCompaction) {
   }
 }
 
+TEST_F(DBCompactionAbortTest, AbortScheduledAutomaticCompactionBeforePick) {
+  // The goal is to cover an automatic compaction that has been scheduled, but
+  // aborts before the worker picks and removes a column family from the
+  // compaction queue. This pins the scheduler bookkeeping invariant that
+  // ResumeAllCompactions() must be able to schedule that still-queued work.
+  constexpr int kCompactionTrigger = 4;
+  constexpr int kNumL0Files = kCompactionTrigger + 1;
+  Options options = GetOptionsWithStats();
+  options.level0_file_num_compaction_trigger = kCompactionTrigger;
+  options.max_background_compactions = 1;
+  options.max_subcompactions = 1;
+  options.disable_auto_compactions = false;
+  Reopen(options);
+
+  SyncPointAbortHelper helper("BackgroundCallCompaction:0");
+  helper.Setup(dbfull());
+
+  PopulateData(/*num_files=*/kNumL0Files, /*keys_per_file=*/100,
+               /*value_size=*/1000);
+  helper.CleanupAndWait();
+
+  const uint64_t compact_write_bytes_before_resume =
+      stats_->getTickerCount(COMPACT_WRITE_BYTES);
+  dbfull()->ResumeAllCompactions();
+  ASSERT_OK(dbfull()->TEST_WaitForCompact());
+
+  ASSERT_GT(stats_->getTickerCount(COMPACT_WRITE_BYTES),
+            compact_write_bytes_before_resume);
+  ASSERT_LT(NumTableFilesAtLevel(0), kNumL0Files);
+  VerifyDataIntegrity(/*num_keys=*/100);
+}
+
 TEST_F(DBCompactionAbortTest, AbortAndVerifyNoOutputFiles) {
   Options options = CurrentOptions();
   options.level0_file_num_compaction_trigger = 4;

db/db_etc2_test.cc

@@ -35,7 +35,7 @@ namespace ROCKSDB_NAMESPACE {
 
 class DBTest2 : public DBTestBase {
  public:
-  DBTest2() : DBTestBase("db_test2", /*env_do_fsync=*/true) {}
+  DBTest2() : DBTestBase("db_etc2_test", /*env_do_fsync=*/true) {}
 };
 
 TEST_F(DBTest2, OpenForReadOnly) {

db/db_etc3_test.cc

@@ -4,6 +4,9 @@
 //  (found in the LICENSE.Apache file in the root directory).
 
 #include "db/db_test_util.h"
+#include "rocksdb/convenience.h"
+#include "rocksdb/metadata.h"
+#include "rocksdb/sst_file_writer.h"
 
 namespace ROCKSDB_NAMESPACE {
 
@@ -43,50 +46,98 @@ TEST_F(DBEtc3Test, AutoTuneManifestSize) {
   ASSERT_EQ(DBOptions{}.max_manifest_space_amp_pct, 500);
 
   Options options = CurrentOptions();
-  ASSERT_OK(db_->SetOptions({{"level0_file_num_compaction_trigger", "20"}}));
+  // Test setup: create many flushed files. Keep level compaction semantics so
+  // DeleteFilesInRanges can remove non-L0 files, but prevent automatic
+  // compactions and write stalls from adding unrelated behavior.
+  options.disable_auto_compactions = true;
+  options.level0_slowdown_writes_trigger = 100;
+  options.level0_stop_writes_trigger = 200;
 
-  // Use large column family names to essentially control the amount of payload
-  // data needed for the manifest file. Drop manifest entries don't include the
-  // CF name so are small.
+  // Test strategy: use large column family names to control the rough amount
+  // of payload added to the MANIFEST. Drop manifest entries do not include the
+  // CF name, so they are small.
+  //
+  // Most CF helper calls piggy-back a background manifest write so the main
+  // auto-tuning phases continue to test the unrelaxed background threshold
+  // even though CF manipulation itself is foreground. Phase-specific foreground
+  // checks disable that piggy-backed background write.
   uint64_t prev_manifest_num = 0, cur_manifest_num = 0;
   std::deque<ColumnFamilyHandle*> handles;
   int counter = 5;
-  auto AddCfFn = [&]() {
+  auto UpdateManifestNumsFrom = [&](uint64_t before_manifest_num) {
+    prev_manifest_num = before_manifest_num;
+    cur_manifest_num = dbfull()->TEST_Current_Manifest_FileNo();
+  };
+  auto BackgroundManifestWriteFn = [&]() {
+    uint64_t before_manifest_num = cur_manifest_num;
+    ASSERT_OK(Put("x", std::to_string(counter++)));
+    ASSERT_OK(Flush());
+    UpdateManifestNumsFrom(before_manifest_num);
+  };
+  auto AddCfFn = [&](bool include_background_manifest_write = true) {
+    uint64_t before_manifest_num = cur_manifest_num;
     std::string name = "cf" + std::to_string(counter++);
     name.resize(1000, 'a');
     ASSERT_OK(db_->CreateColumnFamily(options, name, &handles.emplace_back()));
-    prev_manifest_num = cur_manifest_num;
     cur_manifest_num = dbfull()->TEST_Current_Manifest_FileNo();
+    if (include_background_manifest_write) {
+      BackgroundManifestWriteFn();
+    }
+    UpdateManifestNumsFrom(before_manifest_num);
   };
-  auto DropCfFn = [&]() {
+  auto DropCfFn = [&](bool include_background_manifest_write = true) {
+    uint64_t before_manifest_num = cur_manifest_num;
     ASSERT_OK(db_->DropColumnFamily(handles.front()));
     ASSERT_OK(db_->DestroyColumnFamilyHandle(handles.front()));
     handles.pop_front();
-    prev_manifest_num = cur_manifest_num;
-    cur_manifest_num = dbfull()->TEST_Current_Manifest_FileNo();
-  };
-  auto TrivialManifestWriteFn = [&]() {
-    ASSERT_OK(Put("x", std::to_string(counter++)));
-    ASSERT_OK(Flush());
-    prev_manifest_num = cur_manifest_num;
     cur_manifest_num = dbfull()->TEST_Current_Manifest_FileNo();
+    if (include_background_manifest_write) {
+      BackgroundManifestWriteFn();
+    }
+    UpdateManifestNumsFrom(before_manifest_num);
   };
 
+  // ---- Phase 1: foreground threshold relaxation is bounded ----
+  //
+  // Foreground operations should only get about 25% extra headroom, not an
+  // unbounded threshold. With 1000-byte CF names and a 3000-byte normal limit,
+  // the relaxed limit should allow the first four foreground-only CF additions
+  // but require rotation on the fifth.
+  options.max_manifest_file_size = 3000;
+  options.max_manifest_space_amp_pct = 0;
+  DestroyAndReopen(options);
+  cur_manifest_num = dbfull()->TEST_Current_Manifest_FileNo();
+  prev_manifest_num = cur_manifest_num;
+  for (int i = 1; i <= 4; ++i) {
+    AddCfFn(/*include_background_manifest_write=*/false);
+    ASSERT_EQ(prev_manifest_num, cur_manifest_num);
+  }
+  AddCfFn(/*include_background_manifest_write=*/false);
+  ASSERT_LT(prev_manifest_num, cur_manifest_num);
+  while (!handles.empty()) {
+    ASSERT_OK(db_->DestroyColumnFamilyHandle(handles.front()));
+    handles.pop_front();
+  }
+
+  // ---- Phase 2: no auto-tuning means frequent rotation ----
+  //
   options.max_manifest_file_size = 1000000;
   options.max_manifest_space_amp_pct = 0;  // no auto-tuning yet
   DestroyAndReopen(options);
+  cur_manifest_num = dbfull()->TEST_Current_Manifest_FileNo();
+  prev_manifest_num = cur_manifest_num;
 
-  // With the generous (minimum) maximum manifest size, should not be rotated
+  // With the generous minimum manifest size, should not be rotated.
   AddCfFn();
   AddCfFn();
   AddCfFn();
   ASSERT_EQ(prev_manifest_num, cur_manifest_num);
 
-  // Change options for small max and (still) no auto-tuning
+  // Lower the minimum while still disabling auto-tuning.
   ASSERT_OK(db_->SetDBOptions({{"max_manifest_file_size", "3000"}}));
 
   // Takes effect on the next manifest write
-  TrivialManifestWriteFn();
+  BackgroundManifestWriteFn();
   ASSERT_LT(prev_manifest_num, cur_manifest_num);
 
   // Now we have to rewrite the whole manifest on each write because the
@@ -97,28 +148,31 @@ TEST_F(DBEtc3Test, AutoTuneManifestSize) {
   ASSERT_LT(prev_manifest_num, cur_manifest_num);
   AddCfFn();
   ASSERT_LT(prev_manifest_num, cur_manifest_num);
-  TrivialManifestWriteFn();
+  BackgroundManifestWriteFn();
   ASSERT_LT(prev_manifest_num, cur_manifest_num);
 
+  // ---- Phase 3: auto-tuning raises the background threshold ----
+  //
   // Enabling auto-tuning should fix this, immediately for next manifest writes.
-  // This will allow up to double-ish the size of the compacted manifest,
-  // which last should have been 4000 + some bytes.
+  // This will allow up to roughly double the size of the compacted manifest,
+  // which now includes CF entries plus the piggy-backed background writes.
   ASSERT_EQ(handles.size(), 4U);
-  ASSERT_OK(db_->SetDBOptions({{"max_manifest_space_amp_pct", "105"}}));
+  ASSERT_OK(db_->SetDBOptions({{"max_manifest_space_amp_pct", "95"}}));
 
-  // After 9 CF names should be enough to rotate the manifest
-  for (int i = 1; i <= 5; ++i) {
+  // Auto-tuning lets several more CF names accumulate in the MANIFEST before
+  // the piggy-backed background write crosses the threshold.
+  for (int i = 1; i <= 3; ++i) {
     if ((i % 2) == 1) {
       DropCfFn();
     }
     AddCfFn();
     ASSERT_EQ(prev_manifest_num, cur_manifest_num);
   }
-  TrivialManifestWriteFn();
+  AddCfFn();
   ASSERT_LT(prev_manifest_num, cur_manifest_num);
 
-  // We now have a different last compacted manifest size, should be
-  // able to go beyond 9 CFs named in manifest this time.
+  // We now have a different last compacted manifest size, so the next
+  // threshold should be based on the newly compacted MANIFEST.
   ASSERT_EQ(handles.size(), 6U);
 
   DropCfFn();
@@ -128,11 +182,10 @@ TEST_F(DBEtc3Test, AutoTuneManifestSize) {
     AddCfFn();
     ASSERT_EQ(prev_manifest_num, cur_manifest_num);
   }
-  // We've written 10 named CFs to the manifest. We should be able to
-  // dynamically change the auto-tuning still based on the last "compacted"
-  // manifest size of 7000 + some bytes.
+  // We should be able to dynamically change the auto-tuning still based on
+  // the last "compacted" manifest size.
   ASSERT_OK(db_->SetDBOptions({{"max_manifest_space_amp_pct", "51"}}));
-  TrivialManifestWriteFn();
+  BackgroundManifestWriteFn();
   ASSERT_LT(prev_manifest_num, cur_manifest_num);
   // And the "compacted" manifest size has reset again, so should be changed
   // again sooner.
@@ -141,16 +194,129 @@ TEST_F(DBEtc3Test, AutoTuneManifestSize) {
     AddCfFn();
     ASSERT_EQ(prev_manifest_num, cur_manifest_num);
   }
-  // Enough for manifest change
-  AddCfFn();
+
+  // ---- Phase 4: foreground operations use relaxed threshold ----
+  //
+  // The current MANIFEST is now large enough for the next background manifest
+  // write to rotate it, but still small enough for foreground operations to use
+  // their 25% extra headroom. Assert that each foreground operation stays on
+  // the same MANIFEST, then verify the next background write rotates.
+  const std::string external_files_dir = dbname_ + "/external_files";
+  ASSERT_OK(env_->CreateDirIfMissing(external_files_dir));
+  auto WriteExternalSstFile = [&](const std::string& file_name,
+                                  const std::string& key,
+                                  const std::string& value) {
+    const std::string file_path = external_files_dir + "/" + file_name;
+    Status ignored = env_->DeleteFile(file_path);
+    ignored.PermitUncheckedError();
+
+    SstFileWriter sst_file_writer(EnvOptions(), options);
+    ASSERT_OK(sst_file_writer.Open(file_path));
+    ASSERT_OK(sst_file_writer.Put(key, value));
+    ASSERT_OK(sst_file_writer.Finish());
+  };
+  auto IngestExternalFileForegroundManifestWriteFn =
+      [&](std::string* ingested_key) {
+        uint64_t before_manifest_num = cur_manifest_num;
+        const std::string key = "z" + std::to_string(counter++);
+        const std::string value = "v" + std::to_string(counter++);
+        const std::string file_name =
+            "ingest_" + std::to_string(counter++) + ".sst";
+        const std::string file_path = external_files_dir + "/" + file_name;
+        WriteExternalSstFile(file_name, key, value);
+
+        ASSERT_OK(
+            db_->IngestExternalFile({file_path}, IngestExternalFileOptions()));
+        ASSERT_EQ(value, Get(key));
+        *ingested_key = key;
+        Status ignored = env_->DeleteFile(file_path);
+        ignored.PermitUncheckedError();
+        UpdateManifestNumsFrom(before_manifest_num);
+      };
+  auto CreateColumnFamilyWithImportForegroundManifestWriteFn = [&]() {
+    uint64_t before_manifest_num = cur_manifest_num;
+    const std::string cf_name = "import_cf" + std::to_string(counter++);
+    const std::string key = "import" + std::to_string(counter++);
+    const std::string value = "v" + std::to_string(counter++);
+    const std::string file_name =
+        "import_" + std::to_string(counter++) + ".sst";
+    const std::string file_path = external_files_dir + "/" + file_name;
+    WriteExternalSstFile(file_name, key, value);
+
+    LiveFileMetaData file_metadata;
+    file_metadata.name = file_name;
+    file_metadata.db_path = external_files_dir;
+    file_metadata.smallest_seqno = 0;
+    file_metadata.largest_seqno = 0;
+    file_metadata.level = 0;
+    ExportImportFilesMetaData metadata;
+    metadata.files.push_back(file_metadata);
+    metadata.db_comparator_name = options.comparator->Name();
+
+    ColumnFamilyHandle* import_handle = nullptr;
+    ASSERT_OK(db_->CreateColumnFamilyWithImport(options, cf_name,
+                                                ImportColumnFamilyOptions(),
+                                                metadata, &import_handle));
+    ASSERT_NE(import_handle, nullptr);
+    handles.push_back(import_handle);
+
+    std::string result;
+    ASSERT_OK(db_->Get(ReadOptions(), import_handle, key, &result));
+    ASSERT_EQ(value, result);
+    Status ignored = env_->DeleteFile(file_path);
+    ignored.PermitUncheckedError();
+    UpdateManifestNumsFrom(before_manifest_num);
+  };
+  auto DeleteFilesInRangesForegroundManifestWriteFn =
+      [&](const std::string& key) {
+        uint64_t before_manifest_num = cur_manifest_num;
+        const std::string limit = key + "\xff";
+        std::vector<RangeOpt> ranges;
+        ranges.emplace_back(key, limit);
+        ASSERT_OK(DeleteFilesInRanges(db_.get(), db_->DefaultColumnFamily(),
+                                      ranges.data(), ranges.size(),
+                                      /*include_end=*/false));
+
+        std::string result;
+        ASSERT_TRUE(db_->Get(ReadOptions(), key, &result).IsNotFound());
+        UpdateManifestNumsFrom(before_manifest_num);
+      };
+
+  // Column family manipulation.
+  AddCfFn(/*include_background_manifest_write=*/false);
+  ASSERT_EQ(prev_manifest_num, cur_manifest_num);
+
+  // SetOptions should not write to the MANIFEST. If that regresses, the write
+  // should be treated as a background write and rotate here.
+  {
+    ASSERT_OK(db_->SetOptions({{"level0_slowdown_writes_trigger", "101"}}));
+    UpdateManifestNumsFrom(cur_manifest_num);
+  }
+  ASSERT_EQ(prev_manifest_num, cur_manifest_num);
+
+  // External file ingestion.
+  std::string ingested_key;
+  IngestExternalFileForegroundManifestWriteFn(&ingested_key);
+  ASSERT_EQ(prev_manifest_num, cur_manifest_num);
+
+  // Imported column family creation.
+  CreateColumnFamilyWithImportForegroundManifestWriteFn();
+  ASSERT_EQ(prev_manifest_num, cur_manifest_num);
+
+  // Physical file deletion by range.
+  DeleteFilesInRangesForegroundManifestWriteFn(ingested_key);
+  ASSERT_EQ(prev_manifest_num, cur_manifest_num);
+
+  // Background flush.
+  BackgroundManifestWriteFn();
   ASSERT_LT(prev_manifest_num, cur_manifest_num);
 
-  // ---- Verify persisted compacted manifest size survives close/reopen ----
+  // ---- Phase 5: persisted compacted manifest size survives close/reopen ----
   // Close with CFs still live. Reopen with reuse_manifest_on_open so the
   // manifest is NOT rewritten from scratch. The persisted compacted size
   // should be loaded and used for auto-tuning.
-  // At this point we have 7 CF handles plus default.
-  ASSERT_EQ(handles.size(), 7U);
+  // At this point we have 8 CF handles plus default.
+  ASSERT_EQ(handles.size(), 8U);
 
   // Collect CF names for reopen, then release handles (Close needs this)
   std::vector<std::string> cf_names = {"default"};
@@ -163,18 +329,16 @@ TEST_F(DBEtc3Test, AutoTuneManifestSize) {
   handles.clear();
 
   Close();
-  // Use a large max_manifest_file_size so the reused manifest (which is
-  // already ~10KB) does NOT trigger rotation on the first few writes.
-  // Auto-tuning with the persisted compacted size (~5KB) at 200% amp
-  // gives a tuned threshold of ~15KB. Without persistence, the threshold
-  // would be max(max_manifest_file_size, 0 * anything) =
-  // max_manifest_file_size.
+  // Use a max_manifest_file_size below the reused manifest size. Auto-tuning
+  // with the persisted compacted size at 200% amp keeps the tuned threshold
+  // high enough. Without persistence, the threshold would be
+  // max(max_manifest_file_size, 0 * anything) = max_manifest_file_size.
   //
-  // We set max_manifest_file_size to two values to distinguish:
-  // - 3000: if persisted compacted size is NOT loaded, tuned = 3000,
-  //   and the first AddCf will rotate (manifest is already ~10KB > 3000)
-  // - With persisted compacted size loaded, tuned = max(3000, 5000*3) = 15000,
-  //   so no rotation until we exceed 15KB
+  // With max_manifest_file_size set to 3000, missing persisted compacted size
+  // would keep tuned = 3000 and the first AddCf would rotate because the
+  // reused manifest is already larger. With persisted compacted size loaded,
+  // the tuned threshold is based on the compacted size, so no rotation until
+  // the manifest grows well beyond the minimum.
   options.max_manifest_file_size = 3000;
   options.max_manifest_space_amp_pct = 200;
   options.reuse_manifest_on_open = true;
@@ -186,11 +350,11 @@ TEST_F(DBEtc3Test, AutoTuneManifestSize) {
   // during reopen, because the persisted compacted size keeps the tuned
   // threshold high enough. Without persistence, last_compacted = 0, so
   // tuned = max_manifest_file_size = 3000, and the first LogAndApply
-  // during Open rotates the manifest because it's already ~10KB > 3000.
+  // during Open rotates the manifest because it is already larger than 3000.
   ASSERT_EQ(manifest_num_before_reopen, cur_manifest_num);
 
-  // Adding CFs should still not trigger rotation because the tuned
-  // threshold (~15KB) exceeds the current manifest size (~10KB + adds).
+  // Adding CFs should still not trigger rotation because the tuned threshold
+  // from the persisted compacted size exceeds the current manifest size.
   for (int i = 1; i <= 4; ++i) {
     AddCfFn();
     ASSERT_EQ(prev_manifest_num, cur_manifest_num);
@@ -202,7 +366,7 @@ TEST_F(DBEtc3Test, AutoTuneManifestSize) {
 
   // Wrap up
   while (!handles.empty()) {
-    DropCfFn();
+    DropCfFn(/*include_background_manifest_write=*/false);
   }
 }
 

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);
 
@@ -5853,6 +5855,7 @@ Status DBImpl::DeleteFilesInRanges(ColumnFamilyHandle* column_family,
   }
 
   VersionEdit edit;
+  edit.MarkForegroundOperation();
   std::set<FileMetaData*> deleted_files;
   JobContext job_context(next_job_id_.fetch_add(1), true);
   {
@@ -6963,7 +6966,9 @@ Status DBImpl::IngestExternalFiles(
         assert(!cfd->IsDropped());
         cfds_to_commit.push_back(cfd);
         autovector<VersionEdit*> edit_list;
-        edit_list.push_back(ingestion_jobs[i].edit());
+        auto* edit = ingestion_jobs[i].edit();
+        edit->MarkForegroundOperation();
+        edit_list.push_back(edit);
         edit_lists.push_back(edit_list);
         ++num_entries;
       }
@@ -6978,7 +6983,6 @@ Status DBImpl::IngestExternalFiles(
       }
       status =
           versions_->LogAndApply(cfds_to_commit, read_options, write_options,
-
                                  edit_lists, &mutex_, directories_.GetDbDir());
       // It is safe to update VersionSet last seqno here after LogAndApply since
       // LogAndApply persists last sequence number from VersionEdits,
@@ -7111,6 +7115,7 @@ Status DBImpl::CreateColumnFamilyWithImport(
 
   SuperVersionContext dummy_sv_ctx(/* create_superversion */ true);
   VersionEdit dummy_edit;
+  dummy_edit.MarkForegroundOperation();
   uint64_t next_file_number = 0;
   std::unique_ptr<std::list<uint64_t>::iterator> pending_output_elem;
   {
@@ -7168,9 +7173,10 @@ Status DBImpl::CreateColumnFamilyWithImport(
 
       // Install job edit [Mutex will be unlocked here]
       if (status.ok()) {
-        status = versions_->LogAndApply(cfd, read_options, write_options,
-                                        import_job.edit(), &mutex_,
-                                        directories_.GetDbDir());
+        auto* edit = import_job.edit();
+        edit->MarkForegroundOperation();
+        status = versions_->LogAndApply(cfd, read_options, write_options, edit,
+                                        &mutex_, directories_.GetDbDir());
         if (status.ok()) {
           InstallSuperVersionForConfigChange(cfd, &sv_context);
         }
@@ -7601,6 +7607,7 @@ Status DBImpl::ReserveFileNumbersBeforeIngestion(
   // reuse the file number that has already assigned to the internal file,
   // and this will overwrite the external file. To protect the external
   // file, we have to make sure the file number will never being reused.
+  dummy_edit.MarkForegroundOperation();
   s = versions_->LogAndApply(cfd, read_options, write_options, &dummy_edit,
                              &mutex_, directories_.GetDbDir());
   if (s.ok()) {

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;
   }
@@ -4067,6 +4067,13 @@ Status DBImpl::BackgroundCompaction(bool* made_progress,
       status = Status::ShutdownInProgress();
     } else if (compaction_aborted_.load(std::memory_order_acquire) > 0) {
       status = Status::Incomplete(Status::SubCode::kCompactionAborted);
+      if (!is_prepicked) {
+        // This automatic compaction was scheduled from compaction_queue_, but
+        // abort happened before PickCompactionFromQueue() could remove the
+        // queued CF. Restore the unscheduled count so ResumeAllCompactions()
+        // can schedule the still-queued work.
+        unscheduled_compactions_++;
+      }
     } else if (is_manual &&
                manual_compaction->canceled.load(std::memory_order_acquire)) {
       status = Status::Incomplete(Status::SubCode::kManualCompactionPaused);
@@ -4074,10 +4081,14 @@ Status DBImpl::BackgroundCompaction(bool* made_progress,
   } else {
     status = error_handler_.GetBGError();
     // If we get here, it means a hard error happened after this compaction
-    // was scheduled by MaybeScheduleFlushOrCompaction(), but before it got
-    // a chance to execute. Since we didn't pop a cfd from the compaction
-    // queue, increment unscheduled_compactions_
-    unscheduled_compactions_++;
+    // was scheduled by MaybeScheduleFlushOrCompaction(), but before it got a
+    // chance to execute. Since non-prepicked work consumed an unscheduled
+    // compaction credit without popping a cfd from the compaction queue,
+    // restore the credit here. Prepicked work was scheduled directly and did
+    // not consume unscheduled_compactions_.
+    if (!is_prepicked) {
+      unscheduled_compactions_++;
+    }
   }
 
   if (!status.ok()) {

db/db_iter_test.cc

@@ -216,12 +216,19 @@ class TestIterator : public InternalIterator {
   bool IsKeyPinned() const override { return true; }
   bool IsValuePinned() const override { return true; }
 
+  void Prepare(const MultiScanArgs* /*scan_opts*/) override {
+    ++prepare_call_count_;
+  }
+
+  size_t prepare_call_count() const { return prepare_call_count_; }
+
  private:
   bool initialized_;
   bool valid_;
   size_t sequence_number_;
   size_t iter_;
   size_t steps_ = 0;
+  size_t prepare_call_count_ = 0;
 
   InternalKeyComparator cmp;
   std::vector<std::pair<std::string, std::string>> data_;
@@ -243,6 +250,53 @@ class DBIteratorTest : public testing::Test {
   DBIteratorTest() : env_(Env::Default()) {}
 };
 
+TEST_F(DBIteratorTest, PrepareForwardsValidatedScanRanges) {
+  Options options;
+  ImmutableOptions ioptions = ImmutableOptions(options);
+  MutableCFOptions mutable_cf_options = MutableCFOptions(options);
+  TestIterator* internal_iter = new TestIterator(BytewiseComparator());
+  internal_iter->AddPut("a", "val_a");
+  internal_iter->AddPut("b", "val_b");
+  internal_iter->Finish();
+
+  ReadOptions ro;
+  std::unique_ptr<Iterator> db_iter(DBIter::NewIter(
+      env_, ro, ioptions, mutable_cf_options, BytewiseComparator(),
+      internal_iter, nullptr /* version */, 10 /* sequence */,
+      nullptr /* read_callback */, /*active_mem=*/nullptr));
+
+  MultiScanArgs scan_opts(BytewiseComparator());
+  scan_opts.insert("a", "c");
+
+  db_iter->Prepare(scan_opts);
+  ASSERT_OK(db_iter->status());
+  EXPECT_EQ(internal_iter->prepare_call_count(), 1);
+}
+
+TEST_F(DBIteratorTest, PrepareDoesNotForwardInvalidScanRanges) {
+  Options options;
+  ImmutableOptions ioptions = ImmutableOptions(options);
+  MutableCFOptions mutable_cf_options = MutableCFOptions(options);
+  TestIterator* internal_iter = new TestIterator(BytewiseComparator());
+  internal_iter->AddPut("a", "val_a");
+  internal_iter->AddPut("b", "val_b");
+  internal_iter->Finish();
+
+  ReadOptions ro;
+  std::unique_ptr<Iterator> db_iter(DBIter::NewIter(
+      env_, ro, ioptions, mutable_cf_options, BytewiseComparator(),
+      internal_iter, nullptr /* version */, 10 /* sequence */,
+      nullptr /* read_callback */, /*active_mem=*/nullptr));
+
+  MultiScanArgs scan_opts(BytewiseComparator());
+  scan_opts.insert("b", "d");
+  scan_opts.insert("a", "c");
+
+  db_iter->Prepare(scan_opts);
+  EXPECT_TRUE(db_iter->status().IsInvalidArgument());
+  EXPECT_EQ(internal_iter->prepare_call_count(), 0);
+}
+
 TEST_F(DBIteratorTest, DBIteratorPrevNext) {
   Options options;
   ImmutableOptions ioptions = ImmutableOptions(options);

db/db_iterator_test.cc

@@ -4395,6 +4395,71 @@ TEST_P(DBMultiScanIteratorTest, RangeAcrossFiles) {
   iter.reset();
 }
 
+TEST_P(DBMultiScanIteratorTest, SortedRangesSkipIODispatcherSort) {
+  auto options = CurrentOptions();
+  options.target_file_size_base = 100 << 10;
+  options.compaction_style = kCompactionStyleUniversal;
+  options.num_levels = 50;
+  options.compression = kNoCompression;
+  DestroyAndReopen(options);
+
+  auto key = [](int i) {
+    std::stringstream ss;
+    ss << "k" << std::setw(5) << std::setfill('0') << i;
+    return ss.str();
+  };
+
+  auto rnd = Random::GetTLSInstance();
+  for (int i = 0; i < 100; ++i) {
+    ASSERT_OK(Put(key(i), rnd->RandomString(2 << 10)));
+  }
+  ASSERT_OK(Flush());
+
+  ASSERT_OK(db_->CompactRange({}, nullptr, nullptr));
+  ASSERT_EQ(2, NumTableFilesAtLevel(49));
+
+  int sort_count = 0;
+  SyncPoint::GetInstance()->SetCallBack(
+      "IODispatcherImpl::SubmitJob:SortBlockHandles",
+      [&](void* /*arg*/) { ++sort_count; });
+  SyncPoint::GetInstance()->EnableProcessing();
+
+  auto tracking_dispatcher = std::make_shared<TrackingIODispatcher>();
+  std::vector<std::string> key_ranges({key(10), key(90)});
+  MultiScanArgs scan_options(BytewiseComparator());
+  scan_options.io_dispatcher = tracking_dispatcher;
+  scan_options.use_async_io = false;
+  scan_options.insert(key_ranges[0], key_ranges[1]);
+
+  ReadOptions ro;
+  ro.fill_cache = GetParam();
+  ColumnFamilyHandle* cfh = dbfull()->DefaultColumnFamily();
+  std::unique_ptr<MultiScan> iter =
+      dbfull()->NewMultiScan(ro, cfh, scan_options);
+
+  try {
+    int i = 10;
+    for (auto range : *iter) {
+      for (auto it : range) {
+        ASSERT_EQ(it.first.ToString(), key(i));
+        ++i;
+      }
+    }
+    ASSERT_EQ(i, 90);
+  } catch (MultiScanException& ex) {
+    ASSERT_NOK(ex.status());
+    std::cerr << "Iterator returned status " << ex.what();
+    abort();
+  } catch (std::logic_error& ex) {
+    std::cerr << "Iterator returned logic error " << ex.what();
+    abort();
+  }
+
+  ASSERT_GT(tracking_dispatcher->GetReadSets().size(), 0);
+  EXPECT_EQ(sort_count, 0);
+  iter.reset();
+}
+
 TEST_P(DBMultiScanIteratorTest, FailureTest) {
   auto options = CurrentOptions();
   options.compression = kNoCompression;

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/version_edit.h

@@ -978,6 +978,11 @@ class VersionEdit {
 
   bool IsColumnFamilyDrop() const { return is_column_family_drop_; }
 
+  void MarkForegroundOperation() { is_foreground_operation_ = true; }
+  bool IsForegroundOperation() const {
+    return is_foreground_operation_ || IsColumnFamilyManipulation();
+  }
+
   void MarkNoManifestWriteDummy() { is_no_manifest_write_dummy_ = true; }
   bool IsNoManifestWriteDummy() const { return is_no_manifest_write_dummy_; }
 
@@ -1111,6 +1116,7 @@ class VersionEdit {
   // it also includes column family name.
   bool is_column_family_drop_ = false;
   bool is_column_family_add_ = false;
+  bool is_foreground_operation_ = false;
   std::string column_family_name_;
 
   uint32_t remaining_entries_ = 0;

db/version_set.cc

@@ -6111,9 +6111,32 @@ Status VersionSet::ProcessManifestWrites(
 
   uint64_t prev_manifest_file_size = manifest_file_size_;
   assert(pending_manifest_file_number_ == 0);
+  bool has_foreground_operation = false;
+  for (const VersionEdit* e : batch_edits) {
+    if (e->IsForegroundOperation()) {
+      has_foreground_operation = true;
+      break;
+    }
+  }
+  // For MANIFEST write batches with any foreground operation (external file
+  // ingestion/import, DeleteFilesInRange(s), and column family manipulations
+  // like CreateColumnFamily and DropColumnFamily), relax the size limit by 25%
+  // to reduce the likelihood of a user operation blocking on MANIFEST rotation.
+  // Background-only batches (flush/compaction) still rotate at the normal
+  // threshold.
+  // TODO/future: for workloads like atomic-replace ingestion-only, with zero
+  // or few flushes and compactions, it might be nice to trigger background
+  // manifest rotation if we are beyond the soft limit. But the vast majority
+  // of workloads should have plenty of background manifest ops to avoid
+  // foreground rotation.
+  uint64_t enforced_limit = tuned_max_manifest_file_size_;
+  if (has_foreground_operation) {
+    uint64_t new_limit = enforced_limit + enforced_limit / 4;
+    // don't keep in case of overflow
+    enforced_limit = std::max(enforced_limit, new_limit);
+  }
   if (!skip_manifest_write &&
-      (!descriptor_log_ ||
-       prev_manifest_file_size >= tuned_max_manifest_file_size_)) {
+      (!descriptor_log_ || prev_manifest_file_size >= enforced_limit)) {
     TEST_SYNC_POINT("VersionSet::ProcessManifestWrites:BeforeNewManifest");
     new_descriptor_log = true;
   } else {

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/convenience.h

@@ -432,9 +432,33 @@ Status GetOptionsFromString(const ConfigOptions& config_options,
                             const Options& base_options,
                             const std::string& opts_str, Options* new_options);
 
+// StringToMap parses a serialized options string into a map. Each
+// resulting map value is in a self-contained form (it can be embedded
+// directly in a `key=value;` context -- e.g. SetOptions -- without further
+// escaping). Specifically: nested braced values from the input are
+// preserved with their outer braces. Permissive: accepts both braced and
+// unbraced forms; values not requiring braces are returned as-is.
+// Example:
+//   "filter_policy={id=ribbonfilter:10;bloom_before_level=-1};block_size=4096"
+// produces:
+//   {filter_policy -> "{id=ribbonfilter:10;bloom_before_level=-1}",
+//    block_size    -> "4096"}
 Status StringToMap(const std::string& opts_str,
                    std::unordered_map<std::string, std::string>* opts_map);
 
+// MapToString is the inverse of StringToMap: a naive `key=value;` join.
+// Each map value must already be in self-contained form (as returned by
+// StringToMap) -- i.e. simple text or a single balanced `{...}` block.
+// Values from StringToMap satisfy this property, so the round-trip
+//
+//   StringToMap(MapToString(StringToMap(s))) == StringToMap(s)
+//
+// holds. Callers building a map by hand are responsible for ensuring
+// values are self-contained; raw values containing `;` or starting with
+// `{` without matching `}` won't round-trip.
+Status MapToString(const std::unordered_map<std::string, std::string>& opts_map,
+                   std::string* opts_str);
+
 // Request stopping background work, if wait is true wait until it's done
 void CancelAllBackgroundWork(DB* db, bool wait = false);
 

include/rocksdb/io_dispatcher.h

@@ -152,6 +152,11 @@ class BlockBasedTable;
 struct JobOptions {
   uint64_t io_coalesce_threshold = 16 * 1024;
   ReadOptions read_options;
+
+  // True when IOJob::block_handles is already sorted by file offset. Callers
+  // that can guarantee sorted order may set this to let IODispatcher skip
+  // defensive sorting before cache lookup and coalescing.
+  bool block_handles_are_sorted = false;
 };
 
 class IOJob {
@@ -255,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;
@@ -1003,6 +1004,13 @@ struct DBOptions {
   //
   // This option is mutable with SetDBOptions(), taking effect on the next
   // manifest write (e.g. completed DB compaction or flush).
+  //
+  // For MANIFEST write batches containing foreground operations like external
+  // file ingestion/import, DeleteFilesInRange, CreateColumnFamily, and
+  // DropColumnFamily, the effective limit is relaxed by 25% to reduce the
+  // likelihood of user operations blocking on MANIFEST rotation.
+  // Background-only batches (flush and compaction) use the configured or
+  // auto-tuned limit directly.
   uint64_t max_manifest_file_size = 1024 * 1024 * 1024;
 
   // If true, on DB close, read back the entire MANIFEST file and validate
@@ -2379,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/utilities/options_type.h

@@ -453,11 +453,13 @@ class OptionTypeInfo {
                                        const std::string& value, void* addr) {
           std::map<std::string, std::string> map;
           Status s;
+          // Permit `{k1=v1;k2=v2}` (as serialized) or bare `k1=v1;k2=v2`.
+          std::string stripped = OptionTypeInfo::StripOuterBraces(value);
           for (size_t start = 0, end = 0;
-               s.ok() && start < value.size() && end != std::string::npos;
+               s.ok() && start < stripped.size() && end != std::string::npos;
                start = end + 1) {
             std::string token;
-            s = OptionTypeInfo::NextToken(value, item_separator, start, &end,
+            s = OptionTypeInfo::NextToken(stripped, item_separator, start, &end,
                                           &token);
             if (s.ok() && !token.empty()) {
               size_t pos = token.find(kv_separator);
@@ -947,6 +949,14 @@ class OptionTypeInfo {
   static Status NextToken(const std::string& opts, char delimiter, size_t start,
                           size_t* end, std::string* token);
 
+  // Strips a single layer of outer "{" / "}" wrapping iff the leading '{'
+  // pairs with the trailing '}'. This permissively unwraps values that
+  // could legitimately be either braced or bare (e.g. `{a:b:c}` and
+  // `a:b:c` are accepted equivalently by list-style parsers). The check
+  // is done at brace depth: `{a}:{b}` is left as-is because the leading
+  // '{' closes before the end of the string.
+  static std::string StripOuterBraces(const std::string& value);
+
   constexpr static const char* kIdPropName() { return "id"; }
   constexpr static const char* kIdPropSuffix() { return ".id"; }
 
@@ -995,12 +1005,16 @@ Status ParseArray(const ConfigOptions& config_options,
 
   ConfigOptions copy = config_options;
   copy.ignore_unsupported_options = false;
+  // Permit the caller to pass either bare `item1:item2` or wrapped
+  // `{item1:item2}`. Either form parses identically.
+  std::string stripped = OptionTypeInfo::StripOuterBraces(value);
   size_t i = 0, start = 0, end = 0;
-  for (; status.ok() && i < kSize && start < value.size() &&
+  for (; status.ok() && i < kSize && start < stripped.size() &&
          end != std::string::npos;
        i++, start = end + 1) {
     std::string token;
-    status = OptionTypeInfo::NextToken(value, separator, start, &end, &token);
+    status =
+        OptionTypeInfo::NextToken(stripped, separator, start, &end, &token);
     if (status.ok()) {
       status = elem_info.Parse(copy, name, token, &((*result)[i]));
       if (config_options.ignore_unsupported_options &&
@@ -1137,11 +1151,15 @@ Status ParseVector(const ConfigOptions& config_options,
   // object is valid or not.
   ConfigOptions copy = config_options;
   copy.ignore_unsupported_options = false;
+  // Permit the caller to pass either bare `item1:item2` or wrapped
+  // `{item1:item2}`. Either form parses identically.
+  std::string stripped = OptionTypeInfo::StripOuterBraces(value);
   for (size_t start = 0, end = 0;
-       status.ok() && start < value.size() && end != std::string::npos;
+       status.ok() && start < stripped.size() && end != std::string::npos;
        start = end + 1) {
     std::string token;
-    status = OptionTypeInfo::NextToken(value, separator, start, &end, &token);
+    status =
+        OptionTypeInfo::NextToken(stripped, separator, start, &end, &token);
     if (status.ok()) {
       T elem;
       status = elem_info.Parse(copy, name, token, &elem);

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)

options/db_options.cc

@@ -557,11 +557,13 @@ static std::unordered_map<std::string, OptionTypeInfo>
             embedded.ignore_unsupported_options = true;
             std::vector<std::shared_ptr<EventListener>> listeners;
             Status s;
+            // Permit `{item1:item2}` or bare `item1:item2`.
+            std::string stripped = OptionTypeInfo::StripOuterBraces(value);
             for (size_t start = 0, end = 0;
-                 s.ok() && start < value.size() && end != std::string::npos;
+                 s.ok() && start < stripped.size() && end != std::string::npos;
                  start = end + 1) {
               std::string token;
-              s = OptionTypeInfo::NextToken(value, ':', start, &end, &token);
+              s = OptionTypeInfo::NextToken(stripped, ':', start, &end, &token);
               if (s.ok() && !token.empty()) {
                 std::shared_ptr<EventListener> listener;
                 s = EventListener::CreateFromString(embedded, token, &listener);

options/options_helper.cc

@@ -770,6 +770,11 @@ Status StringToMap(const std::string& opts_str,
   // Example:
   //   opts_str = "write_buffer_size=1024;max_write_buffer_number=2;"
   //              "nested_opt={opt1=1;opt2=2};max_bytes_for_level_base=100"
+  //
+  // Each value in the resulting map can be substituted directly into a
+  // `key=value;` context (e.g. SetOptions) and re-parsed without ambiguity:
+  // values that were wrapped in `{...}` in the input keep their wrapping,
+  // so embedded `;` characters stay escaped.
   size_t pos = 0;
   std::string opts = trim(opts_str);
   // If the input string starts and ends with "{...}", strip off the brackets
@@ -790,23 +795,78 @@ Status StringToMap(const std::string& opts_str,
       return Status::InvalidArgument("Empty key found");
     }
 
+    // Locate value start (after '=' and any whitespace).
+    size_t value_start = eq_pos + 1;
+    while (value_start < opts.size() && isspace(opts[value_start])) {
+      ++value_start;
+    }
+
     std::string value;
-    Status s = OptionTypeInfo::NextToken(opts, ';', eq_pos + 1, &pos, &value);
-    if (!s.ok()) {
-      return s;
-    } else {
-      (*opts_map)[key] = value;
-      if (pos == std::string::npos) {
-        break;
-      } else {
-        pos++;
+    if (value_start < opts.size() && opts[value_start] == '{') {
+      // Braced value: extract the entire `{...}` substring INCLUDING braces
+      // so that the value can be re-emitted as-is in a `key=value;` context.
+      int count = 1;
+      size_t brace_pos = value_start + 1;
+      while (brace_pos < opts.size() && count > 0) {
+        if (opts[brace_pos] == '{') {
+          ++count;
+        } else if (opts[brace_pos] == '}') {
+          --count;
+        }
+        if (count > 0) {
+          ++brace_pos;
+        }
       }
+      if (count != 0) {
+        return Status::InvalidArgument(
+            "Mismatched curly braces for nested options");
+      }
+      value = opts.substr(value_start, brace_pos - value_start + 1);
+      pos = brace_pos + 1;
+      // Allow whitespace, then either delimiter or end.
+      while (pos < opts.size() && isspace(opts[pos])) {
+        ++pos;
+      }
+      if (pos < opts.size() && opts[pos] != ';') {
+        return Status::InvalidArgument("Unexpected chars after nested options");
+      }
+    } else {
+      // Non-braced: scan to the next ';'. The value may contain '=' (only
+      // the first '=' separates key from value).
+      size_t end = opts.find(';', value_start);
+      if (end == std::string::npos) {
+        value = trim(opts.substr(value_start));
+        pos = std::string::npos;
+      } else {
+        value = trim(opts.substr(value_start, end - value_start));
+        pos = end;
+      }
+    }
+
+    (*opts_map)[key] = value;
+    if (pos == std::string::npos) {
+      break;
+    } else {
+      pos++;
     }
   }
 
   return Status::OK();
 }
 
+Status MapToString(const std::unordered_map<std::string, std::string>& opts_map,
+                   std::string* opts_str) {
+  assert(opts_str);
+  opts_str->clear();
+  for (const auto& [key, value] : opts_map) {
+    opts_str->append(key);
+    opts_str->append("=");
+    opts_str->append(value);
+    opts_str->append(";");
+  }
+  return Status::OK();
+}
+
 Status GetStringFromDBOptions(std::string* opt_string,
                               const DBOptions& db_options,
                               const std::string& delimiter) {
@@ -1053,6 +1113,32 @@ Status OptionTypeInfo::NextToken(const std::string& opts, char delimiter,
   return Status::OK();
 }
 
+std::string OptionTypeInfo::StripOuterBraces(const std::string& value) {
+  if (value.size() < 2 || value.front() != '{' || value.back() != '}') {
+    return value;
+  }
+  // Verify the leading '{' actually pairs with the trailing '}'.
+  // For `{a:b}` the leading brace closes only at the trailing `}` (one
+  // matching pair, OK to strip). For `{a}:{b}` the leading brace closes
+  // at the first inner `}`, so the leading and trailing braces are
+  // independent -- leave the value alone. We strip at most one layer:
+  // each layer of `{}` corresponds to one level of nesting that the
+  // encoder added for that level, and the typed parser at this level
+  // wants to peel exactly its own layer.
+  int depth = 0;
+  for (size_t i = 0; i + 1 < value.size(); ++i) {
+    if (value[i] == '{') {
+      ++depth;
+    } else if (value[i] == '}') {
+      --depth;
+      if (depth == 0) {
+        return value;  // outer braces are independent
+      }
+    }
+  }
+  return value.substr(1, value.size() - 2);
+}
+
 Status OptionTypeInfo::Parse(const ConfigOptions& config_options,
                              const std::string& opt_name,
                              const std::string& value, void* opt_ptr) const {
@@ -1071,7 +1157,13 @@ Status OptionTypeInfo::Parse(const ConfigOptions& config_options,
       copy.invoke_prepare_options = false;
       void* opt_addr = GetOffset(opt_ptr);
       return parse_func_(copy, opt_name, opt_value, opt_addr);
-    } else if (ParseOptionHelper(GetOffset(opt_ptr), type_, opt_value)) {
+    } else if (ParseOptionHelper(GetOffset(opt_ptr), type_,
+                                 StripOuterBraces(opt_value))) {
+      // Scalar types (int, bool, enum, string, ...). StringToMap now
+      // preserves outer braces in nested values, so a hand-crafted
+      // `key={42}` (or `db_log_dir={/tmp}`) lands here with a wrap
+      // that scalar parsers don't expect; permissively strip one
+      // level so braced scalar input still parses as before.
       return Status::OK();
     } else if (IsConfigurable()) {
       // The option is <config>.<name>

options/options_test.cc

@@ -1938,11 +1938,13 @@ TEST_F(OptionsTest, StringToMapTest) {
   ASSERT_EQ(opts_map["k2"], "");
   ASSERT_TRUE(opts_map.find("k3") != opts_map.end());
   ASSERT_EQ(opts_map["k3"], "");
-  // Regular nested options
+  // Regular nested options. Braces are preserved on values that came in
+  // braced, so the value is self-contained for direct embedding via
+  // `key=value;` and round-trips through StringToMap/MapToString.
   opts_map.clear();
   ASSERT_OK(StringToMap("k1=v1;k2={nk1=nv1;nk2=nv2};k3=v3", &opts_map));
   ASSERT_EQ(opts_map["k1"], "v1");
-  ASSERT_EQ(opts_map["k2"], "nk1=nv1;nk2=nv2");
+  ASSERT_EQ(opts_map["k2"], "{nk1=nv1;nk2=nv2}");
   ASSERT_EQ(opts_map["k3"], "v3");
   // Multi-level nested options
   opts_map.clear();
@@ -1951,34 +1953,35 @@ TEST_F(OptionsTest, StringToMapTest) {
                   "k3={nk1={nnk1={nnnk1=nnnv1;nnnk2;nnnv2}}};k4=v4",
                   &opts_map));
   ASSERT_EQ(opts_map["k1"], "v1");
-  ASSERT_EQ(opts_map["k2"], "nk1=nv1;nk2={nnk1=nnk2}");
-  ASSERT_EQ(opts_map["k3"], "nk1={nnk1={nnnk1=nnnv1;nnnk2;nnnv2}}");
+  ASSERT_EQ(opts_map["k2"], "{nk1=nv1;nk2={nnk1=nnk2}}");
+  ASSERT_EQ(opts_map["k3"], "{nk1={nnk1={nnnk1=nnnv1;nnnk2;nnnv2}}}");
   ASSERT_EQ(opts_map["k4"], "v4");
   // Garbage inside curly braces
   opts_map.clear();
   ASSERT_OK(StringToMap("k1=v1;k2={dfad=};k3={=};k4=v4", &opts_map));
   ASSERT_EQ(opts_map["k1"], "v1");
-  ASSERT_EQ(opts_map["k2"], "dfad=");
-  ASSERT_EQ(opts_map["k3"], "=");
+  ASSERT_EQ(opts_map["k2"], "{dfad=}");
+  ASSERT_EQ(opts_map["k3"], "{=}");
   ASSERT_EQ(opts_map["k4"], "v4");
   // Empty nested options
   opts_map.clear();
   ASSERT_OK(StringToMap("k1=v1;k2={};", &opts_map));
   ASSERT_EQ(opts_map["k1"], "v1");
-  ASSERT_EQ(opts_map["k2"], "");
+  ASSERT_EQ(opts_map["k2"], "{}");
   opts_map.clear();
   ASSERT_OK(StringToMap("k1=v1;k2={{{{}}}{}{}};", &opts_map));
   ASSERT_EQ(opts_map["k1"], "v1");
-  ASSERT_EQ(opts_map["k2"], "{{{}}}{}{}");
-  // With random spaces
+  ASSERT_EQ(opts_map["k2"], "{{{{}}}{}{}}");
+  // With random spaces. Internal whitespace is preserved verbatim now
+  // (we keep the original substring of the value rather than trimming).
   opts_map.clear();
   ASSERT_OK(
       StringToMap("  k1 =  v1 ; k2= {nk1=nv1; nk2={nnk1=nnk2}}  ; "
                   "k3={  {   } }; k4= v4  ",
                   &opts_map));
   ASSERT_EQ(opts_map["k1"], "v1");
-  ASSERT_EQ(opts_map["k2"], "nk1=nv1; nk2={nnk1=nnk2}");
-  ASSERT_EQ(opts_map["k3"], "{   }");
+  ASSERT_EQ(opts_map["k2"], "{nk1=nv1; nk2={nnk1=nnk2}}");
+  ASSERT_EQ(opts_map["k3"], "{  {   } }");
   ASSERT_EQ(opts_map["k4"], "v4");
 
   // Empty key
@@ -2008,6 +2011,165 @@ TEST_F(OptionsTest, StringToMapTest) {
   ASSERT_NOK(StringToMap("k1=v1;k2={{dfdl}adfa}{}", &opts_map));
 }
 
+TEST_F(OptionsTest, EmptyBracedVectorAndBracedScalarTest) {
+  // An empty braced list value (`key={}`) must parse as an empty
+  // collection, not as a vector with one empty element. StringToMap
+  // preserves the braces, so the typed vector parser sees `{}`; its
+  // outer-brace strip needs to handle the size-2 case.
+  ColumnFamilyOptions cf_opts;
+  ConfigOptions cfg;
+  ASSERT_OK(GetColumnFamilyOptionsFromString(
+      cfg, ColumnFamilyOptions(), "compression_per_level={};", &cf_opts));
+  EXPECT_TRUE(cf_opts.compression_per_level.empty());
+
+  // A scalar value wrapped in `{}` (e.g. hand-crafted `key={42}`) must
+  // still parse. StringToMap preserves the braces, so the scalar
+  // dispatch in OptionTypeInfo::Parse permissively strips one outer
+  // layer.
+  DBOptions db_opts;
+  ASSERT_OK(GetDBOptionsFromString(cfg, DBOptions(), "max_open_files={42};",
+                                   &db_opts));
+  EXPECT_EQ(db_opts.max_open_files, 42);
+
+  ASSERT_OK(GetDBOptionsFromString(cfg, DBOptions(),
+                                   "create_if_missing={true};", &db_opts));
+  EXPECT_TRUE(db_opts.create_if_missing);
+}
+
+TEST_F(OptionsTest, MapToStringTest) {
+  using Map = std::unordered_map<std::string, std::string>;
+  std::string s;
+
+  // Simple values.
+  Map simple{{"a", "1"}, {"b", "hello"}};
+  ASSERT_OK(MapToString(simple, &s));
+  Map reparsed;
+  ASSERT_OK(StringToMap(s, &reparsed));
+  ASSERT_EQ(reparsed, simple);
+
+  // A value already in self-contained braced form (as returned by
+  // StringToMap) passes through unchanged. No extra wrapping is needed
+  // for round-trip.
+  Map already_braced{{"k", "{a=b;c=d}"}};
+  ASSERT_OK(MapToString(already_braced, &s));
+  ASSERT_EQ(s, "k={a=b;c=d};");
+  Map reparsed_braced;
+  ASSERT_OK(StringToMap(s, &reparsed_braced));
+  ASSERT_EQ(reparsed_braced, already_braced);
+}
+
+TEST_F(OptionsTest, StringToMapMapToStringRoundTripTest) {
+  using Map = std::unordered_map<std::string, std::string>;
+  // Pull a single entry out of a StringToMap-ed map and re-embed it
+  // directly into a `key=value;` SetOptions-style string. The pulled
+  // value must round-trip without any extra escaping by the caller --
+  // this is the property that lets users compose option strings from
+  // map entries safely.
+  const std::string original =
+      "filter_policy={id=ribbonfilter:10:-1;bloom_before_level=-1;};"
+      "block_size=4096;cache_index_and_filter_blocks=true";
+
+  Map parsed;
+  ASSERT_OK(StringToMap(original, &parsed));
+  // The filter_policy value retains its braces -- it's self-contained.
+  EXPECT_EQ(parsed["filter_policy"],
+            "{id=ribbonfilter:10:-1;bloom_before_level=-1;}");
+  EXPECT_EQ(parsed["block_size"], "4096");
+  EXPECT_EQ(parsed["cache_index_and_filter_blocks"], "true");
+
+  // Pull a single entry out and embed it in a fresh `key=value;`
+  // string. No extra wrapping required by the caller.
+  const std::string embedded =
+      "filter_policy=" + parsed["filter_policy"] + ";other_opt=42";
+  Map reparsed;
+  ASSERT_OK(StringToMap(embedded, &reparsed));
+  EXPECT_EQ(reparsed["filter_policy"], parsed["filter_policy"]);
+  EXPECT_EQ(reparsed["other_opt"], "42");
+
+  // Whole-map round-trip via MapToString.
+  std::string regenerated;
+  ASSERT_OK(MapToString(parsed, &regenerated));
+  Map reparsed2;
+  ASSERT_OK(StringToMap(regenerated, &reparsed2));
+  EXPECT_EQ(reparsed2, parsed);
+}
+
+TEST_F(OptionsTest, FullOptionsStringToMapRoundTripTest) {
+  // Round-trip a populated DBOptions and CFOptions through the
+  // StringToMap -> MapToString path. This catches any custom option
+  // serializer that emits a value containing the StringToMap delimiter
+  // (';') without enclosing it in '{}' -- such a value would survive
+  // StringToMap as a self-contained entry but corrupt subsequent
+  // re-serialization. Existing typed tests
+  // (ColumnFamilyOptionsSerialization, DBOptionsSerialization) cover the
+  // typed Configurable round-trip; this test specifically exercises the
+  // public StringToMap + MapToString path that callers use to compose,
+  // mutate, and re-emit option strings.
+  Random rnd(607);
+  ConfigOptions config_options;
+  config_options.input_strings_escaped = false;
+
+  // ---- ColumnFamilyOptions ----
+  Options options;
+  ColumnFamilyOptions base_cf_opt;
+  base_cf_opt.comparator = test::BytewiseComparatorWithU64TsWrapper();
+  test::RandomInitCFOptions(&base_cf_opt, options, &rnd);
+  std::string cf_str;
+  ASSERT_OK(
+      GetStringFromColumnFamilyOptions(config_options, base_cf_opt, &cf_str));
+
+  std::unordered_map<std::string, std::string> cf_map;
+  ASSERT_OK(StringToMap(cf_str, &cf_map));
+  // Each entry must be embeddable as-is in a `key=value;` context.
+  for (const auto& [k, v] : cf_map) {
+    std::string solo;
+    solo.append(k).append("=").append(v).append(";");
+    std::unordered_map<std::string, std::string> solo_map;
+    ASSERT_OK(StringToMap(solo, &solo_map))
+        << "single-entry round-trip failed for " << k << ": " << v;
+    ASSERT_EQ(solo_map.size(), 1u) << k << "=" << v;
+    ASSERT_EQ(solo_map[k], v) << "single-entry value mismatch for " << k;
+  }
+  // Whole-map MapToString must reproduce a string parseable into a CF
+  // options object equivalent to the original.
+  std::string cf_round;
+  ASSERT_OK(MapToString(cf_map, &cf_round));
+  ColumnFamilyOptions cf_back;
+  ASSERT_OK(GetColumnFamilyOptionsFromString(
+      config_options, ColumnFamilyOptions(), cf_round, &cf_back));
+  ASSERT_OK(RocksDBOptionsParser::VerifyCFOptions(config_options, base_cf_opt,
+                                                  cf_back));
+
+  // ---- DBOptions ----
+  DBOptions base_db_opt;
+  test::RandomInitDBOptions(&base_db_opt, &rnd);
+  std::string db_str;
+  ASSERT_OK(GetStringFromDBOptions(config_options, base_db_opt, &db_str));
+
+  std::unordered_map<std::string, std::string> db_map;
+  ASSERT_OK(StringToMap(db_str, &db_map));
+  for (const auto& [k, v] : db_map) {
+    std::string solo;
+    solo.append(k).append("=").append(v).append(";");
+    std::unordered_map<std::string, std::string> solo_map;
+    ASSERT_OK(StringToMap(solo, &solo_map))
+        << "single-entry round-trip failed for " << k << ": " << v;
+    ASSERT_EQ(solo_map.size(), 1u) << k << "=" << v;
+    ASSERT_EQ(solo_map[k], v) << "single-entry value mismatch for " << k;
+  }
+  std::string db_round;
+  ASSERT_OK(MapToString(db_map, &db_round));
+  DBOptions db_back;
+  ASSERT_OK(
+      GetDBOptionsFromString(config_options, DBOptions(), db_round, &db_back));
+  ASSERT_OK(RocksDBOptionsParser::VerifyDBOptions(config_options, base_db_opt,
+                                                  db_back));
+
+  // RandomInitCFOptions allocates a raw CompactionFilter*; the
+  // ColumnFamilyOptions destructor doesn't own it.
+  delete base_cf_opt.compaction_filter;
+}
+
 TEST_F(OptionsTest, StringToMapRandomTest) {
   std::unordered_map<std::string, std::string> opts_map;
   // Make sure segfault is not hit by semi-random strings
@@ -2028,6 +2190,15 @@ TEST_F(OptionsTest, StringToMapRandomTest) {
         str[pos] = ' ';
         Status s = StringToMap(str, &opts_map);
         ASSERT_TRUE(s.ok() || s.IsInvalidArgument());
+        if (s.ok()) {
+          // Round-trip: StringToMap entries are self-contained, so
+          // MapToString of them must parse back to the same map.
+          std::string regen;
+          ASSERT_OK(MapToString(opts_map, &regen));
+          std::unordered_map<std::string, std::string> reparsed;
+          ASSERT_OK(StringToMap(regen, &reparsed));
+          ASSERT_EQ(reparsed, opts_map) << "regen=" << regen;
+        }
         opts_map.clear();
       }
     }
@@ -2047,8 +2218,22 @@ TEST_F(OptionsTest, StringToMapRandomTest) {
     }
     Status s = StringToMap(str, &opts_map);
     ASSERT_TRUE(s.ok() || s.IsInvalidArgument());
+    if (s.ok()) {
+      std::string regen;
+      ASSERT_OK(MapToString(opts_map, &regen));
+      std::unordered_map<std::string, std::string> reparsed;
+      ASSERT_OK(StringToMap(regen, &reparsed));
+      ASSERT_EQ(reparsed, opts_map) << "regen=" << regen;
+    }
     s = StringToMap("name=" + str, &opts_map);
     ASSERT_TRUE(s.ok() || s.IsInvalidArgument());
+    if (s.ok()) {
+      std::string regen;
+      ASSERT_OK(MapToString(opts_map, &regen));
+      std::unordered_map<std::string, std::string> reparsed;
+      ASSERT_OK(StringToMap(regen, &reparsed));
+      ASSERT_EQ(reparsed, opts_map) << "regen=" << regen;
+    }
     opts_map.clear();
   }
 }

src.mk

@@ -506,6 +506,7 @@ TEST_MAIN_SOURCES =                                                     \
   db/db_clip_test.cc                                                    \
   db/db_dynamic_level_test.cc                                           \
   db/db_encryption_test.cc                                              \
+  db/db_etc2_test.cc                                                    \
   db/db_etc3_test.cc                                                    \
   db/db_flush_test.cc                                                   \
   db/db_follower_test.cc						                                    \
@@ -533,7 +534,6 @@ TEST_MAIN_SOURCES =                                                     \
   db/db_table_properties_test.cc                                        \
   db/db_tailing_iter_test.cc                                            \
   db/db_test.cc                                                         \
-  db/db_test2.cc                                                        \
   db/db_logical_block_size_cache_test.cc                                \
   db/db_universal_compaction_test.cc                                    \
   db/db_wal_test.cc                                                     \

table/block_based/block_based_table_iterator.cc

@@ -1077,6 +1077,9 @@ void BlockBasedTableIterator::Prepare(const MultiScanArgs* multiscan_opts) {
       multiscan_opts->io_coalesce_threshold;
   job->job_options.read_options = read_options_;
   job->job_options.read_options.async_io = multiscan_opts->use_async_io;
+  // CollectBlockHandles walks sorted scan ranges through the table index, whose
+  // key order matches data block offset order.
+  job->job_options.block_handles_are_sorted = true;
 
   std::shared_ptr<ReadSet> read_set;
   // IODispatcher should be provided by DBIter::Prepare() to enable sharing

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/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/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/compression_test.cc

@@ -521,7 +521,7 @@ class PresetCompressionDictTest
       public testing::WithParamInterface<std::tuple<CompressionType, bool>> {
  public:
   PresetCompressionDictTest()
-      : DBTestBase("db_test2", false /* env_do_fsync */),
+      : DBTestBase("compression_test_preset_dict", false /* env_do_fsync */),
         compression_type_(std::get<0>(GetParam())),
         bottommost_(std::get<1>(GetParam())) {}
 

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 {
@@ -44,13 +47,37 @@ struct IODispatcherImplData {
   virtual void ReleaseMemory(size_t bytes) = 0;
 };
 
+#ifndef NDEBUG
+static bool BlockIndicesAreSortedByOffset(
+    const std::vector<BlockHandle>& block_handles,
+    const std::vector<size_t>& block_indices) {
+  uint64_t prev_offset = 0;
+  for (size_t i = 0; i < block_indices.size(); ++i) {
+    if (block_indices[i] >= block_handles.size()) {
+      return false;
+    }
+    const uint64_t current_offset = block_handles[block_indices[i]].offset();
+    if (i > 0 && current_offset < prev_offset) {
+      return false;
+    }
+    prev_offset = current_offset;
+  }
+  return true;
+}
+#endif  // NDEBUG
+
 // Helper function to create and pin a block from a buffer
 // Used by both ReadSet::PollAndProcessAsyncIO and IODispatcherImpl::Impl
 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();
@@ -71,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)
@@ -97,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;
@@ -111,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);
     }
   }
 
@@ -149,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);
   }
 }
 
@@ -171,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();
@@ -200,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();
       }
 
@@ -272,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 {
@@ -320,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
@@ -356,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();
@@ -376,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
@@ -479,7 +637,8 @@ struct IODispatcherImpl::Impl : public IODispatcherImplData,
                         const std::vector<size_t>& block_indices);
 
   // Pre-coalesce blocks into groups, respecting max_group_bytes size limit.
-  // Returns groups ordered by first block index (earlier blocks first).
+  // block_indices must be sorted by block offset.
+  // Returns groups ordered by block offset.
   std::vector<CoalescedPrefetchGroup> PreCoalesceBlocks(
       const std::shared_ptr<IOJob>& job, const std::shared_ptr<ReadSet>& rs,
       const std::vector<size_t>& block_indices, size_t max_group_bytes);
@@ -697,30 +856,43 @@ Status IODispatcherImpl::Impl::SubmitJob(const std::shared_ptr<IOJob>& job,
   for (size_t i = 0; i < job->block_handles.size(); ++i) {
     rs->sorted_block_indices_[i] = i;
   }
-  std::sort(rs->sorted_block_indices_.begin(), rs->sorted_block_indices_.end(),
-            [&job](size_t a, size_t b) {
-              return job->block_handles[a].offset() <
-                     job->block_handles[b].offset();
-            });
+  if (!job->job_options.block_handles_are_sorted) {
+    TEST_SYNC_POINT("IODispatcherImpl::SubmitJob:SortBlockHandles");
+    std::sort(rs->sorted_block_indices_.begin(),
+              rs->sorted_block_indices_.end(), [&job](size_t a, size_t b) {
+                return job->block_handles[a].offset() <
+                       job->block_handles[b].offset();
+              });
+  }
 
-  // Step 1: Check cache and pin cached blocks
+  // Step 1: Check cache and pin cached blocks. Iterate in offset order so all
+  // downstream private helpers can assume block index vectors are already
+  // 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 = 0; i < job->block_handles.size(); ++i) {
-    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);
+      }
     }
   }
 
@@ -816,17 +988,11 @@ void IODispatcherImpl::Impl::PrepareIORequests(
     const std::vector<BlockHandle>& block_handles,
     std::vector<FSReadRequest>* read_reqs,
     std::vector<std::vector<size_t>>* coalesced_block_indices) {
-  // This is necessary because block handles may not be in sorted order
-  std::vector<size_t> sorted_block_indices = block_indices_to_read;
-  std::sort(sorted_block_indices.begin(), sorted_block_indices.end(),
-            [&block_handles](size_t a, size_t b) {
-              return block_handles[a].offset() < block_handles[b].offset();
-            });
-
+  assert(BlockIndicesAreSortedByOffset(block_handles, block_indices_to_read));
   assert(coalesced_block_indices->empty());
   coalesced_block_indices->resize(1);
 
-  for (const auto& block_idx : sorted_block_indices) {
+  for (const auto& block_idx : block_indices_to_read) {
     if (!coalesced_block_indices->back().empty()) {
       // Check if we can coalesce with previous block
       const auto& last_block_handle =
@@ -883,17 +1049,12 @@ std::vector<CoalescedPrefetchGroup> IODispatcherImpl::Impl::PreCoalesceBlocks(
   const auto& block_handles = job->block_handles;
   const uint64_t coalesce_threshold = job->job_options.io_coalesce_threshold;
 
-  // Sort block indices by offset for coalescing
-  std::vector<size_t> sorted_indices = block_indices;
-  std::sort(sorted_indices.begin(), sorted_indices.end(),
-            [&block_handles](size_t a, size_t b) {
-              return block_handles[a].offset() < block_handles[b].offset();
-            });
+  assert(BlockIndicesAreSortedByOffset(block_handles, block_indices));
 
   // Build coalesced groups respecting max_group_bytes
   groups.emplace_back();
 
-  for (size_t idx : sorted_indices) {
+  for (size_t idx : block_indices) {
     size_t block_size = rs->block_sizes_[idx];
 
     // Skip blocks that are individually larger than the memory budget
@@ -951,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]);
 
@@ -964,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();
@@ -980,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
@@ -1021,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;
@@ -1054,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];
@@ -1076,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;

util/string_util.cc

@@ -251,21 +251,22 @@ std::string UnescapeOptionString(const std::string& escaped_string) {
 }
 
 std::string trim(const std::string& str) {
-  if (str.empty()) {
-    return std::string();
-  }
   size_t start = 0;
-  size_t end = str.size() - 1;
-  while (isspace(str[start]) != 0 && start < end) {
+  size_t end = str.size();
+  for (;;) {
+    if (start >= end) {
+      return {};  // all whitespace
+    }
+    if (!isspace(str[start])) {
+      break;  // first non-whitespace found
+    }
     ++start;
   }
-  while (isspace(str[end]) != 0 && start < end) {
+  while (isspace(str[end - 1]) != 0) {
     --end;
+    assert(end > start);
   }
-  if (start <= end) {
-    return str.substr(start, end - start + 1);
-  }
-  return std::string();
+  return str.substr(start, end - start);
 }
 
 bool EndsWith(const std::string& string, const std::string& pattern) {

util/string_util_test.cc

@@ -27,6 +27,31 @@ TEST(StringUtilTest, NumberToHumanString) {
   ASSERT_EQ("-10G", NumberToHumanString(-10000000000));
 }
 
+TEST(StringUtilTest, Trim) {
+  // Empty input.
+  EXPECT_EQ("", trim(""));
+  // No whitespace to strip.
+  EXPECT_EQ("a", trim("a"));
+  EXPECT_EQ("abc", trim("abc"));
+  // Leading whitespace only.
+  EXPECT_EQ("a", trim(" a"));
+  EXPECT_EQ("abc", trim("   abc"));
+  // Trailing whitespace only.
+  EXPECT_EQ("a", trim("a "));
+  EXPECT_EQ("abc", trim("abc   "));
+  // Both ends.
+  EXPECT_EQ("a", trim(" a "));
+  EXPECT_EQ("abc", trim("   abc   "));
+  EXPECT_EQ("a b c", trim("  a b c  "));  // interior whitespace preserved
+  // All-whitespace inputs of various lengths must trim to empty.
+  EXPECT_EQ("", trim(" "));
+  EXPECT_EQ("", trim("  "));
+  EXPECT_EQ("", trim("   "));
+  EXPECT_EQ("", trim("\t"));
+  EXPECT_EQ("", trim("\n"));
+  EXPECT_EQ("", trim(" \t\n\r"));
+}
+
 }  // namespace ROCKSDB_NAMESPACE
 
 int main(int argc, char** argv) {

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(