Update HISTORY.md and version for 11.1.2

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

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

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

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

Reviewed By: xingbowang

Differential Revision: D109622445

fbshipit-source-id: d7be4b20fce86ccb218a63ac6f5b707b316aac79

HISTORY.md

@@ -1,6 +1,35 @@
 # Rocksdb Change Log
 > NOTE: Entries for next release do not go here. Follow instructions in `unreleased_history/README.txt`
 
+## 11.1.2 (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.1.1 (04/10/2026)
+### Public API Changes
+* Changed experimental feature `ExternalTableReader::Get` and `ExternalTableReader::MultiGet` to use `PinnableSlice` instead of `std::string` for output values, enabling zero-copy pinning. This will break existing implementations.
+* Added `SstFileReader::Get` and `SstFileReader::MultiGet` overloads that accept `PinnableSlice`/`std::vector<PinnableSlice>*`, enabling zero-copy reads when the underlying `TableReader` supports pinning.
+* Added block_decompress_count to PerfContext
+
+### Bug Fixes
+* Fix a memory accounting leak in IODispatcher where ReadIndex() moved block values out of ReadSet without releasing the associated prefetch memory, causing subsequent prefetches to be blocked when max_prefetch_memory_bytes was set.
+
+## 11.1.0 (03/23/2026)
+### New Features
+* Add a new option `open_files_async`. The existing behavior is on DB open, we open all sst files and do basic validations. For very large DBs on remote filesystems with many ssts, this may take very long. This option performs these validations instead in the background. Open errors found by this async background task are surfaced as a new background error kAsyncFileOpen.
+* Added `BlockBasedTableOptions::kAuto` index block search type that automatically selects between binary and interpolation search on a per-index-block basis. During SST construction, each index block's key distribution uniformity is analyzed using the coefficient of variation of key gaps, and index blocks with uniform keys use interpolation search while others fall back to binary search. The uniformity threshold is configurable via `BlockBasedTableOptions::uniform_cv_threshold` (default: 0.2).
+* Introduced enforce_write_buffer_manager_during_recovery option to allow WriteBufferManager to be enforced during WAL recovery. (Default: true)
+* Add `memtable_batch_lookup_optimization` option to use batch lookup optimization for memtable MultiGet. For skip list memtables, after each key lookup, the search path is cached and reused for the next key, reducing per-key cost from O(log N) to O(log d) where d is the distance between consecutive keys. Benchmarks show ~7% improvement in memtable-resident MultiGet throughput.
+* Added `BlockBasedTableOptions::PrepopulateBlockCache::kFlushAndCompaction` to prepopulate the block cache during both flush and compaction. Compaction-warmed blocks are inserted at `BOTTOM` priority (vs `LOW` for flush) so they are evicted first under cache pressure. Recommended only for use cases where most or all of the database is expected to reside in cache (e.g., tiered or remote storage where the working set fits in cache).
+* Added new mutable DB option `verify_manifest_content_on_close` (default: false). When enabled, on DB close the MANIFEST file is read back and all records are validated (CRC checksums and logical content). If corruption is detected, a fresh MANIFEST is written from in-memory state.
+
+### Behavior Changes
+* num_reads_sampled now factors in re-seeks and next/prev() on file iterators for files in L1+. next/prev() is discounted by 64x compared to seek due to being a much cheaper call.
+* Remote compaction workers now skip WAL recovery when opening the secondary DB instance, since only the LSM state from MANIFEST is needed for compaction. This reduces I/O and speeds up remote compaction startup.
+
+### Bug Fixes
+* Fix a bug in round-robin compaction that missed selecting input files that are needed to guarantee data correctness and cause crashing in debug builds or silent data corruption in release builds for Get().
+
 ## 11.0.0 (02/23/2026)
 ### New Features
 * Added support for storing wide-column entity column values in blob files. When `min_blob_size` is configured, large column values in wide-column entities will be stored in blob files, reducing SST file size and improving read performance.

db/db_impl/db_impl.cc

@@ -169,6 +169,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_)),
@@ -614,7 +615,7 @@ Status DBImpl::CloseHelper() {
   // manifest file), it is not able to identify live files correctly. As a
   // result, all "live" files can get deleted by accident. However, corrupted
   // manifest is recoverable by RepairDB().
-  if (opened_successfully_) {
+  if (opened_successfully_ && !read_only_) {
     JobContext job_context(next_job_id_.fetch_add(1));
     FindObsoleteFiles(&job_context, true);
 

db/db_impl/db_impl.h

@@ -1367,6 +1367,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

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

db/db_test.cc

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

db_stress_tool/db_stress_common.h

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

db_stress_tool/db_stress_gflags.cc

@@ -1609,4 +1609,10 @@ DEFINE_bool(use_multiscan, false,
 DEFINE_bool(multiscan_use_async_io, false,
             "If set, enable async_io for MultiScan operations.");
 
+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 "
+              "used for prefetching data blocks across all concurrent "
+              "MultiScan ReadSets.");
+
 #endif  // GFLAGS

db_stress_tool/db_stress_test_base.cc

@@ -29,6 +29,7 @@
 #include "options/options_parser.h"
 #include "rocksdb/convenience.h"
 #include "rocksdb/filter_policy.h"
+#include "rocksdb/io_dispatcher.h"
 #include "rocksdb/secondary_cache.h"
 #include "rocksdb/sst_file_manager.h"
 #include "rocksdb/table_properties.h"
@@ -1715,6 +1716,14 @@ Status StressTest::TestMultiScan(ThreadState* thread,
       FLAGS_multiscan_use_async_io &&
       CheckFSFeatureSupport(options_.env->GetFileSystem().get(),
                             FSSupportedOps::kAsyncIO);
+  std::shared_ptr<IODispatcher> io_dispatcher;
+  if (FLAGS_multiscan_max_prefetch_memory_bytes > 0) {
+    IODispatcherOptions io_opts;
+    io_opts.max_prefetch_memory_bytes =
+        FLAGS_multiscan_max_prefetch_memory_bytes;
+    io_dispatcher.reset(NewIODispatcher(io_opts));
+    scan_opts.io_dispatcher = io_dispatcher;
+  }
   start_key_strs.reserve(num_scans);
   end_key_strs.reserve(num_scans);
 

include/rocksdb/external_table.h

@@ -115,14 +115,14 @@ class ExternalTableReader {
   // Point lookup the given key and return its value
   virtual Status Get(const ReadOptions& read_options, const Slice& key,
                      const SliceTransform* prefix_extractor,
-                     std::string* value) = 0;
+                     PinnableSlice* value) = 0;
 
   // Point lookup the given vector of keys and return the values, as well
   // as status of each individual lookup in statuses.
   virtual void MultiGet(const ReadOptions& read_options,
                         const std::vector<Slice>& keys,
                         const SliceTransform* prefix_extractor,
-                        std::vector<std::string>* values,
+                        std::vector<PinnableSlice>* values,
                         std::vector<Status>* statuses) = 0;
 
   // Allocate and return the contents of the properties block. If the builder

include/rocksdb/perf_context.h

@@ -106,7 +106,8 @@ struct PerfContextBase {
   uint64_t compressed_sec_cache_compressed_bytes;
 
   uint64_t block_checksum_time;    // total nanos spent on block checksum
-  uint64_t block_decompress_time;  // total nanos spent on block decompression
+  uint64_t block_decompress_time;   // total nanos spent on block decompression
+  uint64_t block_decompress_count;  // total number of block decompressions
 
   uint64_t get_read_bytes;       // bytes for vals returned by Get
   uint64_t multiget_read_bytes;  // bytes for vals returned by MultiGet

include/rocksdb/sst_file_reader.h

@@ -35,6 +35,19 @@ class SstFileReader {
                                const std::vector<Slice>& keys,
                                std::vector<std::string>* values);
 
+  // MultiGet variant that returns PinnableSlice values, enabling zero-copy
+  // when the underlying TableReader supports pinning.
+  std::vector<Status> MultiGet(const ReadOptions& options,
+                               const std::vector<Slice>& keys,
+                               std::vector<PinnableSlice>* values);
+
+  // Point lookup a single key from the SST.
+  Status Get(const ReadOptions& options, const Slice& key, std::string* value);
+
+  // Point lookup variant that returns a PinnableSlice.
+  Status Get(const ReadOptions& options, const Slice& key,
+             PinnableSlice* value);
+
   // Returns a new iterator over the table contents as a raw table iterator,
   // a.k.a a `TableIterator`that iterates all point data entries in the table
   // including logically invisible entries like delete entries.

include/rocksdb/version.h

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

monitoring/perf_context.cc

@@ -73,6 +73,7 @@ struct PerfContextByLevelInt {
   defCmd(compressed_sec_cache_compressed_bytes)    \
   defCmd(block_checksum_time)                      \
   defCmd(block_decompress_time)                    \
+  defCmd(block_decompress_count)                   \
   defCmd(get_read_bytes)                           \
   defCmd(multiget_read_bytes)                      \
   defCmd(iter_read_bytes)                          \

table/block_based/block_based_table_reader_test.cc

@@ -1624,6 +1624,76 @@ TEST_P(BlockBasedTableReaderMultiScanTest, MultiScanUnpinPreviousBlocks) {
   }
 }
 
+// Regression test for assertion failure when re-seeking to an already-exhausted
+// scan range. This can happen when MergingIterator re-seeks a child iterator
+// after all scan ranges have been consumed (e.g., due to range tombstone
+// adjustments in other levels). The bug was that SeekToBlockIdx() set
+// valid_=true without bounds checking, while cur_idx_ was past
+// block_handles_.size().
+TEST_P(BlockBasedTableReaderMultiScanTest, MultiScanReseekAfterExhaustion) {
+  std::vector<std::pair<std::string, std::string>> kv =
+      BlockBasedTableReaderBaseTest::GenerateKVMap(
+          10 /* num_block */, true /* mixed_with_human_readable_string_value */,
+          comparator_->timestamp_size(), same_key_diff_ts_, comparator_);
+  std::string table_name = "BlockBasedTableReaderTest_ReseekAfterExhaustion" +
+                           CompressionTypeToString(compression_type_);
+  ImmutableOptions ioptions(options_);
+  CreateTable(table_name, ioptions, compression_type_, kv,
+              compression_parallel_threads_, compression_dict_bytes_);
+
+  std::unique_ptr<BlockBasedTable> table;
+  FileOptions foptions;
+  foptions.use_direct_reads = use_direct_reads_;
+  InternalKeyComparator comparator(options_.comparator);
+  NewBlockBasedTableReader(foptions, ioptions, comparator, table_name, &table,
+                           true /* bool prefetch_index_and_filter_in_cache */,
+                           nullptr /* status */, persist_udt_);
+
+  ReadOptions read_opts;
+  std::unique_ptr<InternalIterator> iter;
+  iter.reset(table->NewIterator(
+      read_opts, options_.prefix_extractor.get(), /*arena=*/nullptr,
+      /*skip_filters=*/false, TableReaderCaller::kUncategorized));
+
+  // Set up two scan ranges covering blocks 0-4 and 5-9
+  MultiScanArgs scan_options(BytewiseComparator());
+  scan_options.insert(ExtractUserKey(kv[0].first),
+                      ExtractUserKey(kv[5 * kEntriesPerBlock - 1].first));
+  scan_options.insert(ExtractUserKey(kv[5 * kEntriesPerBlock].first),
+                      ExtractUserKey(kv[10 * kEntriesPerBlock - 1].first));
+
+  iter->Prepare(&scan_options);
+
+  // Seek to range 1 and iterate through all blocks
+  iter->Seek(kv[0].first);
+  ASSERT_TRUE(iter->Valid());
+  ASSERT_OK(iter->status());
+  while (iter->Valid()) {
+    iter->Next();
+  }
+
+  // Re-seek to the first range's start key after range 1 is exhausted.
+  // The forward-only check in MultiScanIndexIterator rejects this seek,
+  // so the iterator remains in the scan_range_exhausted state (valid but
+  // out-of-bound, signaling the caller to seek to the next range).
+  iter->Seek(kv[0].first);
+
+  // Seek to range 2 and iterate through all blocks
+  iter->Seek(kv[5 * kEntriesPerBlock].first);
+  while (iter->Valid()) {
+    iter->Next();
+  }
+  // Now all scan ranges are fully exhausted.
+
+  // Re-seek to the last range's start key. This simulates what happens when
+  // MergingIterator re-seeks a child due to range tombstone key adjustment.
+  // Before the fix, this would trigger:
+  //   assert(cur_idx_ < block_handles_.size()) in value()
+  // because SeekToBlockIdx set valid_=true with cur_idx_ past bounds.
+  iter->Seek(kv[5 * kEntriesPerBlock].first);
+  ASSERT_FALSE(iter->Valid());
+}
+
 // Test that fs_prefetch_support flag is correctly initialized during table
 // construction based on filesystem capabilities
 TEST_P(BlockBasedTableReaderTest, FSPrefetchSupportInitializedCorrectly) {

table/block_based/multi_scan_index_iterator.cc

@@ -143,6 +143,12 @@ void MultiScanIndexIterator::Seek(const Slice& target) {
     // max_sequential_skip_in_iterations can trigger a reseek on the start
     // key of a scan range, even though we're already past cur_scan_start_idx
     size_t block_idx = std::max(cur_scan_start_idx, cur_idx_);
+    if (block_idx >= cur_scan_end_idx) {
+      // cur_idx_ has advanced past this range's blocks (e.g., due to a
+      // reseek after all ranges were exhausted). Treat as exhausted.
+      SetExhausted();
+      return;
+    }
     SeekToBlockIdx(block_idx);
   }
 }

table/external_table.cc

@@ -8,6 +8,7 @@
 #include "logging/logging.h"
 #include "rocksdb/table.h"
 #include "table/block_based/block.h"
+#include "table/get_context.h"
 #include "table/internal_iterator.h"
 #include "table/meta_blocks.h"
 #include "table/table_builder.h"
@@ -233,10 +234,30 @@ class ExternalTableReaderAdapter : public TableReader {
 
   size_t ApproximateMemoryUsage() const override { return 0; }
 
-  Status Get(const ReadOptions&, const Slice&, GetContext*,
-             const SliceTransform*, bool = false) override {
-    return Status::NotSupported(
-        "Get() not supported on external file iterator");
+  Status Get(const ReadOptions& read_options, const Slice& key,
+             GetContext* get_context, const SliceTransform* prefix_extractor,
+             bool /*skip_filters*/ = false) override {
+    ParsedInternalKey parsed_key;
+    Status s = ParseInternalKey(key, &parsed_key, /*log_err_key=*/false);
+    if (!s.ok()) {
+      return s;
+    }
+
+    PinnableSlice value;
+    s = reader_->Get(read_options, parsed_key.user_key, prefix_extractor,
+                     &value);
+    if (!s.ok()) {
+      if (s.IsNotFound()) {
+        return Status::OK();
+      }
+      return s;
+    }
+
+    ParsedInternalKey found_key(parsed_key.user_key, 0, ValueType::kTypeValue);
+    bool matched = false;
+    get_context->SaveValue(found_key, value, &matched, &s,
+                           value.IsPinned() ? &value : nullptr);
+    return s;
   }
 
   Status VerifyChecksum(const ReadOptions& /*ro*/, TableReaderCaller /*caller*/,

table/format.cc

@@ -712,6 +712,7 @@ Status DecompressBlockData(Decompressor::Args& args, Decompressor& decompressor,
              args.compressed_data.size());
   RecordTick(ioptions.stats, BYTES_DECOMPRESSED_TO, out_contents->data.size());
   RecordTick(ioptions.stats, NUMBER_BLOCK_DECOMPRESSED);
+  PERF_COUNTER_ADD(block_decompress_count, 1);
 
   TEST_SYNC_POINT_CALLBACK("DecompressBlockData:TamperWithReturnValue",
                            static_cast<void*>(&s));

table/sst_file_reader.cc

@@ -83,12 +83,15 @@ Status SstFileReader::Open(const std::string& file_path) {
   return s;
 }
 
-std::vector<Status> SstFileReader::MultiGet(const ReadOptions& roptions,
-                                            const std::vector<Slice>& keys,
-                                            std::vector<std::string>* values) {
+std::vector<Status> SstFileReader::MultiGet(
+    const ReadOptions& roptions, const std::vector<Slice>& keys,
+    std::vector<PinnableSlice>* values) {
   const auto num_keys = keys.size();
   std::vector<Status> statuses(num_keys, Status::OK());
-  std::vector<PinnableSlice> pin_values(num_keys);
+  values->resize(num_keys);
+  for (size_t i = 0; i < num_keys; ++i) {
+    (*values)[i].Reset();
+  }
 
   auto r = rep_.get();
   const Comparator* user_comparator =
@@ -101,8 +104,7 @@ std::vector<Status> SstFileReader::MultiGet(const ReadOptions& roptions,
   autovector<MergeContext, MultiGetContext::MAX_BATCH_SIZE> merge_ctx;
   sorted_keys.resize(num_keys);
   for (size_t i = 0; i < num_keys; ++i) {
-    PinnableSlice* val = &pin_values[i];
-    val->Reset();
+    PinnableSlice* val = &(*values)[i];
     merge_ctx.emplace_back();
     key_context.emplace_back(nullptr, keys[i], val, nullptr,
                              nullptr /* timestamp */, &statuses[i]);
@@ -139,14 +141,12 @@ std::vector<Status> SstFileReader::MultiGet(const ReadOptions& roptions,
                             r->moptions.prefix_extractor.get(),
                             false /* skip filters */);
 
-  values->resize(num_keys);
   for (size_t i = 0; i < num_keys; ++i) {
     get_ctx[i].ReportCounters();
 
     if (statuses[i].ok()) {
       switch (get_ctx[i].State()) {
         case GetContext::kFound:
-          (*values)[i].assign(pin_values[i].data(), pin_values[i].size());
           break;
         case GetContext::kNotFound:
         case GetContext::kDeleted:
@@ -166,6 +166,75 @@ std::vector<Status> SstFileReader::MultiGet(const ReadOptions& roptions,
   return statuses;
 }
 
+std::vector<Status> SstFileReader::MultiGet(const ReadOptions& roptions,
+                                            const std::vector<Slice>& keys,
+                                            std::vector<std::string>* values) {
+  std::vector<PinnableSlice> pin_values;
+  std::vector<Status> statuses = MultiGet(roptions, keys, &pin_values);
+  values->resize(keys.size());
+  for (size_t i = 0; i < keys.size(); ++i) {
+    if (statuses[i].ok()) {
+      (*values)[i].assign(pin_values[i].data(), pin_values[i].size());
+    }
+  }
+  return statuses;
+}
+
+Status SstFileReader::Get(const ReadOptions& roptions, const Slice& key,
+                          PinnableSlice* value) {
+  auto r = rep_.get();
+  value->Reset();
+
+  const Comparator* user_comparator =
+      r->ioptions.internal_comparator.user_comparator();
+  Statistics* statistics = r->ioptions.stats;
+
+  Status status;
+  MergeContext merge_context;
+  SequenceNumber max_covering_tombstone_seq = 0;
+  GetContext get_ctx(user_comparator, r->ioptions.merge_operator.get(),
+                     nullptr /* logger */, statistics, GetContext::kNotFound,
+                     key, value, nullptr /* columns */, nullptr /* timestamp */,
+                     nullptr /* value_found */, &merge_context, true,
+                     &max_covering_tombstone_seq, r->ioptions.clock);
+
+  status = r->table_reader->Get(
+      roptions, InternalKey(key, kMaxSequenceNumber, kTypeValue).Encode(),
+      &get_ctx, r->moptions.prefix_extractor.get(), false /* skip_filters */);
+
+  get_ctx.ReportCounters();
+
+  if (status.ok()) {
+    switch (get_ctx.State()) {
+      case GetContext::kFound:
+        break;
+      case GetContext::kNotFound:
+      case GetContext::kDeleted:
+        status = Status::NotFound();
+        break;
+      case GetContext::kMerge:
+        status = Status::MergeInProgress();
+        break;
+      case GetContext::kCorrupt:
+      case GetContext::kUnexpectedBlobIndex:
+      case GetContext::kMergeOperatorFailed:
+        status = Status::Corruption();
+        break;
+    }
+  }
+  return status;
+}
+
+Status SstFileReader::Get(const ReadOptions& roptions, const Slice& key,
+                          std::string* value) {
+  PinnableSlice pin_value;
+  Status s = Get(roptions, key, &pin_value);
+  if (s.ok()) {
+    value->assign(pin_value.data(), pin_value.size());
+  }
+  return s;
+}
+
 Iterator* SstFileReader::NewIterator(const ReadOptions& roptions) {
   assert(roptions.io_activity == Env::IOActivity::kUnknown);
   auto r = rep_.get();

table/sst_file_reader_test.cc

@@ -768,37 +768,33 @@ TEST_F(SstFileReaderTableIteratorTest, UserDefinedTimestampsEnabled) {
   Close();
 }
 
-class SstFileReaderTableMultiGetTest : public DBTestBase {
+class SstFileReaderTableGetTest : public DBTestBase,
+                                  public testing::WithParamInterface<bool> {
  public:
-  SstFileReaderTableMultiGetTest()
-      : DBTestBase("sst_file_reader_table_multi_get_test",
+  SstFileReaderTableGetTest()
+      : DBTestBase("sst_file_reader_table_get_test",
                    /*env_do_fsync=*/false) {}
 
-  void VerifyTableEntry(Iterator* iter, const std::string& user_key,
-                        ValueType value_type,
-                        std::optional<std::string> expected_value,
-                        bool backward_iteration = false) {
-    ASSERT_TRUE(iter->Valid());
-    ASSERT_TRUE(iter->status().ok());
-    ParsedInternalKey pikey;
-    ASSERT_OK(ParseInternalKey(iter->key(), &pikey, /*log_err_key=*/false));
-    ASSERT_EQ(pikey.user_key, user_key);
-    ASSERT_EQ(pikey.type, value_type);
-    if (expected_value.has_value()) {
-      ASSERT_EQ(iter->value(), expected_value.value());
-    }
-    if (!backward_iteration) {
-      iter->Next();
+  bool UseMultiGet() const { return GetParam(); }
+
+  std::vector<Status> DoGet(SstFileReader& reader,
+                            const std::vector<Slice>& keys,
+                            std::vector<std::string>* values) {
+    if (UseMultiGet()) {
+      return reader.MultiGet(ReadOptions(), keys, values);
     } else {
-      iter->Prev();
+      values->resize(keys.size());
+      std::vector<Status> statuses(keys.size());
+      for (size_t i = 0; i < keys.size(); ++i) {
+        statuses[i] = reader.Get(ReadOptions(), keys[i], &(*values)[i]);
+      }
+      return statuses;
     }
   }
 };
 
-TEST_F(SstFileReaderTableMultiGetTest, Basic) {
+TEST_P(SstFileReaderTableGetTest, Basic) {
   Options options = CurrentOptions();
-  const Comparator* ucmp = BytewiseComparator();
-  options.comparator = ucmp;
   options.disable_auto_compactions = true;
   options.merge_operator = MergeOperators::CreateStringAppendOperator();
   options.statistics = CreateDBStatistics();
@@ -823,7 +819,7 @@ TEST_F(SstFileReaderTableMultiGetTest, Basic) {
 
   std::vector<LiveFileMetaData> files;
   dbfull()->GetLiveFilesMetaData(&files);
-  ASSERT_TRUE(files.size() == 1);
+  ASSERT_EQ(files.size(), 1);
   ASSERT_TRUE(files[0].level == 0);
   std::string file_name = files[0].directory + "/" + files[0].relative_filename;
 
@@ -833,25 +829,30 @@ TEST_F(SstFileReaderTableMultiGetTest, Basic) {
 
   ASSERT_OK(options.statistics->Reset());
 
-  std::vector<Slice> keys;
+  std::vector<Slice> keys = {"fo1", "foo", "baz",
+                             "bar", "aaa", "zzz_not_in_sst"};
   std::vector<std::string> values;
+  auto statuses = DoGet(reader, keys, &values);
 
-  keys.emplace_back("fo1");
-  keys.emplace_back("foo");
-  keys.emplace_back("baz");
-  keys.emplace_back("bar");
-  keys.emplace_back("aaa");
-  auto statuses = reader.MultiGet(ReadOptions(), keys, &values);
-  ASSERT_TRUE(statuses[0].IsNotFound())
-      << "Failed: status=" << statuses[0].ToString() << " val=" << values[0];
-  ASSERT_TRUE(statuses[1].IsNotFound())
-      << "Failed: status=" << statuses[0].ToString() << " val=" << values[1];
-  ASSERT_TRUE(statuses[2].ok());
-  ASSERT_TRUE(statuses[3].ok());
-  ASSERT_EQ("val3", values[2]);
-  ASSERT_EQ("val2", values[3]);
-  ASSERT_TRUE(statuses[4].ok());
-  ASSERT_EQ("val4,val5", values[4]);
+  // Non-existent key returns NotFound
+  ASSERT_TRUE(statuses[0].IsNotFound());
+
+  // Deleted key returns NotFound
+  ASSERT_TRUE(statuses[1].IsNotFound());
+
+  // Found keys
+  ASSERT_OK(statuses[2]);
+  ASSERT_EQ(values[2], "val3");
+
+  ASSERT_OK(statuses[3]);
+  ASSERT_EQ(values[3], "val2");
+
+  // Merged key
+  ASSERT_OK(statuses[4]);
+  ASSERT_EQ(values[4], "val4,val5");
+
+  // Bloom filter filtered key
+  ASSERT_TRUE(statuses[5].IsNotFound());
 
   uint64_t cache_hits = options.statistics->getTickerCount(BLOCK_CACHE_HIT);
   uint64_t cache_misses = options.statistics->getTickerCount(BLOCK_CACHE_MISS);
@@ -863,6 +864,9 @@ TEST_F(SstFileReaderTableMultiGetTest, Basic) {
   Close();
 }
 
+INSTANTIATE_TEST_CASE_P(SingleAndMulti, SstFileReaderTableGetTest,
+                        testing::Bool());
+
 }  // namespace ROCKSDB_NAMESPACE
 
 int main(int argc, char** argv) {

table/table_test.cc

@@ -7045,10 +7045,10 @@ class ExternalTableTest : public DBTestBase {
 
     Status Get(const ReadOptions& /*read_options*/, const Slice& key,
                const SliceTransform* /*prefix_extractor*/,
-               std::string* value) override {
+               PinnableSlice* value) override {
       auto iter = kv_map_.find(key.ToString());
       if (iter != kv_map_.end()) {
-        value->assign(iter->second);
+        value->PinSelf(iter->second);
         return Status::OK();
       }
       return Status::NotFound();
@@ -7057,7 +7057,7 @@ class ExternalTableTest : public DBTestBase {
     void MultiGet(const ReadOptions& read_options,
                   const std::vector<Slice>& keys,
                   const SliceTransform* prefix_extractor,
-                  std::vector<std::string>* values,
+                  std::vector<PinnableSlice>* values,
                   std::vector<Status>* statuses) override {
       values->resize(keys.size());
       statuses->resize(keys.size());
@@ -7091,6 +7091,38 @@ class ExternalTableTest : public DBTestBase {
     bool support_property_block_;
   };
 
+  // A reader that pins values from its internal buffer, exercising the
+  // zero-copy path in ExternalTableReaderAdapter::Get().
+  class PinnedDummyExternalTableReader : public DummyExternalTableReader {
+   public:
+    using DummyExternalTableReader::DummyExternalTableReader;
+
+    Status Get(const ReadOptions& /*read_options*/, const Slice& key,
+               const SliceTransform* /*prefix_extractor*/,
+               PinnableSlice* value) override {
+      auto it = pinned_data_.find(key.ToString());
+      if (it != pinned_data_.end()) {
+        Slice s(it->second);
+        value->PinSlice(s, &PinCleanup, &pin_cleanup_count_, nullptr);
+        return Status::OK();
+      }
+      return Status::NotFound();
+    }
+
+    void SetPinnedData(const std::map<std::string, std::string>& data) {
+      pinned_data_ = data;
+    }
+
+    int pin_cleanup_count() const { return pin_cleanup_count_; }
+
+   private:
+    static void PinCleanup(void* arg1, void* /*arg2*/) {
+      (*static_cast<int*>(arg1))++;
+    }
+    std::map<std::string, std::string> pinned_data_;
+    int pin_cleanup_count_ = 0;
+  };
+
   class DummyExternalTableBuilder : public ExternalTableBuilder {
    public:
     explicit DummyExternalTableBuilder(const std::string& file_path,
@@ -7163,6 +7195,37 @@ class ExternalTableTest : public DBTestBase {
    private:
     bool support_property_block_;
   };
+
+  class PinnedDummyExternalTableFactory : public ExternalTableFactory {
+   public:
+    const char* Name() const override {
+      return "PinnedDummyExternalTableFactory";
+    }
+
+    Status NewTableReader(
+        const ReadOptions& /*read_options*/, const std::string& file_path,
+        const ExternalTableOptions& /*topts*/,
+        std::unique_ptr<ExternalTableReader>* table_reader) const override {
+      auto* reader =
+          new PinnedDummyExternalTableReader(file_path,
+                                             /*support_property_block=*/true);
+      last_reader_ = reader;
+      table_reader->reset(reader);
+      return Status::OK();
+    }
+
+    ExternalTableBuilder* NewTableBuilder(
+        const ExternalTableBuilderOptions& /*opts*/,
+        const std::string& file_path, FSWritableFile* file) const override {
+      return new DummyExternalTableBuilder(file_path, file,
+                                           /*support_property_block=*/true);
+    }
+
+    PinnedDummyExternalTableReader* last_reader() const { return last_reader_; }
+
+   private:
+    mutable PinnedDummyExternalTableReader* last_reader_ = nullptr;
+  };
 };
 
 TEST_F(ExternalTableTest, BasicTest) {
@@ -7200,11 +7263,11 @@ TEST_F(ExternalTableTest, BasicTest) {
   iter->Next();
   ASSERT_FALSE(iter->Valid());
 
-  std::string val;
+  PinnableSlice val;
   ASSERT_OK(reader->Get({}, "foo", nullptr, &val));
   ASSERT_EQ(val, "bar");
 
-  std::vector<std::string> vals;
+  std::vector<PinnableSlice> vals;
   std::vector<Status> statuses;
   reader->MultiGet({}, {"foo", "bar"}, nullptr, &vals, &statuses);
   ASSERT_EQ(vals.size(), 2);
@@ -7232,22 +7295,169 @@ TEST_F(ExternalTableTest, SstReaderTest) {
   std::unique_ptr<SstFileWriter> writer;
   writer.reset(new SstFileWriter(EnvOptions(), options));
   ASSERT_OK(writer->Open(ingest_file));
-  ASSERT_OK(writer->Put("foo", "bar"));
+  ASSERT_OK(writer->Put("a", "val_a"));
+  ASSERT_OK(writer->Put("b", "val_b"));
+  ASSERT_OK(writer->Put("c", "val_c"));
   ASSERT_OK(writer->Finish());
   writer.reset();
 
   std::unique_ptr<SstFileReader> reader(new SstFileReader(options));
   ASSERT_OK(reader->Open(ingest_file));
 
+  // Test iterator
   ReadOptions ro;
   std::unique_ptr<Iterator> iter(reader->NewIterator(ro));
   ASSERT_NE(iter, nullptr);
-  iter->Seek("foo");
+  iter->Seek("a");
   ASSERT_TRUE(iter->Valid() && iter->status().ok());
-  ASSERT_EQ(iter->value(), "bar");
+  ASSERT_EQ(iter->value(), "val_a");
+  iter->Next();
+  ASSERT_TRUE(iter->Valid());
+  ASSERT_EQ(iter->value(), "val_b");
+  iter->Next();
+  ASSERT_TRUE(iter->Valid());
+  ASSERT_EQ(iter->value(), "val_c");
   iter->Next();
   ASSERT_FALSE(iter->Valid());
   ASSERT_TRUE(iter->status().ok());
+
+  // Test MultiGet
+  std::vector<Slice> keys = {"a", "b", "missing", "c"};
+  std::vector<std::string> values;
+  std::vector<Status> statuses = reader->MultiGet(ReadOptions(), keys, &values);
+  ASSERT_EQ(values.size(), keys.size());
+  ASSERT_EQ(statuses.size(), keys.size());
+  ASSERT_OK(statuses[0]);
+  ASSERT_EQ(values[0], "val_a");
+  ASSERT_OK(statuses[1]);
+  ASSERT_EQ(values[1], "val_b");
+  ASSERT_TRUE(statuses[2].IsNotFound());
+  ASSERT_OK(statuses[3]);
+  ASSERT_EQ(values[3], "val_c");
+}
+
+TEST_F(ExternalTableTest, PinnedGetTest) {
+  if (encrypted_env_) {
+    ROCKSDB_GTEST_SKIP("Test requires non-encrypted environment");
+    return;
+  }
+  Options options = GetDefaultOptions();
+  auto factory = std::make_shared<PinnedDummyExternalTableFactory>();
+  options.table_factory = NewExternalTableFactory(factory);
+  Reopen(options);
+
+  std::string ingest_file = dbname_ + "/test.immutabledb";
+
+  std::unique_ptr<SstFileWriter> writer;
+  writer.reset(new SstFileWriter(EnvOptions(), options));
+  ASSERT_OK(writer->Open(ingest_file));
+  ASSERT_OK(writer->Put("key1", "val1"));
+  ASSERT_OK(writer->Put("key2", "val2"));
+  ASSERT_OK(writer->Finish());
+  writer.reset();
+
+  IngestExternalFileOptions ifo;
+  ASSERT_OK(db_->IngestExternalFile({ingest_file}, ifo));
+  ASSERT_NE(factory->last_reader(), nullptr);
+
+  factory->last_reader()->SetPinnedData(
+      {{"key1", "pinned_val1"}, {"key2", "pinned_val2"}});
+
+  PinnableSlice pinnable;
+  ASSERT_OK(
+      db_->Get(ReadOptions(), db_->DefaultColumnFamily(), "key1", &pinnable));
+  ASSERT_EQ(pinnable.ToString(), "pinned_val1");
+  ASSERT_TRUE(pinnable.IsPinned());
+  pinnable.Reset();
+
+  ASSERT_OK(
+      db_->Get(ReadOptions(), db_->DefaultColumnFamily(), "key2", &pinnable));
+  ASSERT_EQ(pinnable.ToString(), "pinned_val2");
+  ASSERT_TRUE(pinnable.IsPinned());
+  pinnable.Reset();
+
+  // Verify cleanup ran for both Gets
+  ASSERT_EQ(factory->last_reader()->pin_cleanup_count(), 2);
+
+  // Verify NotFound still works
+  Status s =
+      db_->Get(ReadOptions(), db_->DefaultColumnFamily(), "missing", &pinnable);
+  ASSERT_TRUE(s.IsNotFound());
+
+  // Test MultiGet with PinnableSlice to exercise the batched pin path
+  const size_t num_keys = 3;
+  std::array<Slice, num_keys> mg_keys = {Slice("key1"), Slice("missing"),
+                                         Slice("key2")};
+  std::array<PinnableSlice, num_keys> mg_values;
+  std::array<Status, num_keys> mg_statuses;
+  db_->MultiGet(ReadOptions(), db_->DefaultColumnFamily(), num_keys,
+                mg_keys.data(), mg_values.data(), mg_statuses.data());
+
+  ASSERT_OK(mg_statuses[0]);
+  ASSERT_EQ(mg_values[0].ToString(), "pinned_val1");
+  ASSERT_TRUE(mg_values[0].IsPinned());
+
+  ASSERT_TRUE(mg_statuses[1].IsNotFound());
+
+  ASSERT_OK(mg_statuses[2]);
+  ASSERT_EQ(mg_values[2].ToString(), "pinned_val2");
+  ASSERT_TRUE(mg_values[2].IsPinned());
+
+  // Reset PinnableSlices to trigger cleanups
+  for (auto& v : mg_values) {
+    v.Reset();
+  }
+  ASSERT_EQ(factory->last_reader()->pin_cleanup_count(), 4);
+}
+
+TEST_F(ExternalTableTest, SstReaderPinnableMultiGetTest) {
+  if (encrypted_env_) {
+    ROCKSDB_GTEST_SKIP("Test requires non-encrypted environment");
+    return;
+  }
+  Options options = GetDefaultOptions();
+  std::string dbname =
+      test::PerThreadDBPath("sst_reader_pinnable_multiget_test");
+  std::string sst_file = dbname + "/test.sst";
+  ASSERT_OK(options.env->CreateDirIfMissing(dbname));
+
+  std::unique_ptr<SstFileWriter> writer(
+      new SstFileWriter(EnvOptions(), options));
+  ASSERT_OK(writer->Open(sst_file));
+  ASSERT_OK(writer->Put("a", "val_a"));
+  ASSERT_OK(writer->Put("b", "val_b"));
+  ASSERT_OK(writer->Put("c", "val_c"));
+  ASSERT_OK(writer->Finish());
+  writer.reset();
+
+  std::unique_ptr<SstFileReader> reader(new SstFileReader(options));
+  ASSERT_OK(reader->Open(sst_file));
+
+  // Test PinnableSlice MultiGet
+  std::vector<Slice> keys = {"a", "b", "missing", "c"};
+  std::vector<PinnableSlice> values;
+  std::vector<Status> statuses = reader->MultiGet(ReadOptions(), keys, &values);
+  ASSERT_EQ(values.size(), keys.size());
+  ASSERT_EQ(statuses.size(), keys.size());
+  ASSERT_OK(statuses[0]);
+  ASSERT_EQ(values[0].ToString(), "val_a");
+  ASSERT_OK(statuses[1]);
+  ASSERT_EQ(values[1].ToString(), "val_b");
+  ASSERT_TRUE(statuses[2].IsNotFound());
+  ASSERT_OK(statuses[3]);
+  ASSERT_EQ(values[3].ToString(), "val_c");
+
+  // Verify std::string MultiGet wrapper still works
+  std::vector<std::string> str_values;
+  statuses = reader->MultiGet(ReadOptions(), keys, &str_values);
+  ASSERT_EQ(str_values.size(), keys.size());
+  ASSERT_OK(statuses[0]);
+  ASSERT_EQ(str_values[0], "val_a");
+  ASSERT_OK(statuses[1]);
+  ASSERT_EQ(str_values[1], "val_b");
+  ASSERT_TRUE(statuses[2].IsNotFound());
+  ASSERT_OK(statuses[3]);
+  ASSERT_EQ(str_values[3], "val_c");
 }
 
 TEST_F(ExternalTableTest, ExternalFileChecksumTest) {

tools/db_crashtest.py

@@ -1272,6 +1272,9 @@ def finalize_and_sanitize(src_params):
         # LevelIterator multiscan currently relies on num_entries and num_range_deletions,
         # which are not updated if skip_stats_update_on_db_open is true
         dest_params["skip_stats_update_on_db_open"] = 0
+        dest_params["multiscan_max_prefetch_memory_bytes"] = random.choice(
+            [0, 0, 64 * 1024, 256 * 1024]
+        )
 
     # open_files_async requires skip_stats_update_on_db_open to avoid
     # synchronous I/O in UpdateAccumulatedStats during DB open

unreleased_history/behavior_changes/num_reads_sampled

@@ -1 +0,0 @@
-num_reads_sampled now factors in re-seeks and next/prev() on file iterators for files in L1+. next/prev() is discounted by 64x compared to seek due to being a much cheaper call. 

unreleased_history/behavior_changes/skip_wal_recovery_remote_compaction

@@ -1 +0,0 @@
-Remote compaction workers now skip WAL recovery when opening the secondary DB instance, since only the LSM state from MANIFEST is needed for compaction. This reduces I/O and speeds up remote compaction startup.

unreleased_history/bug_fixes/fix_rr_duplicate_files.md

@@ -1 +0,0 @@
-Fix a bug in round-robin compaction that missed selecting input files that are needed to guarantee data correctness and cause crashing in debug builds or silent data corruption in release builds for Get().

unreleased_history/new_features/async_file_open

@@ -1 +0,0 @@
-Add a new option `open_files_async`. The existing behavior is on DB open, we open all sst files and do basic validations. For very large DBs on remote filesystems with many ssts, this may take very long. This option performs these validations instead in the background. Open errors found by this async background task are surfaced as a new background error kAsyncFileOpen.

unreleased_history/new_features/auto_interpolation_search.md

@@ -1 +0,0 @@
-Added `BlockBasedTableOptions::kAuto` index block search type that automatically selects between binary and interpolation search on a per-index-block basis. During SST construction, each index block's key distribution uniformity is analyzed using the coefficient of variation of key gaps, and index blocks with uniform keys use interpolation search while others fall back to binary search. The uniformity threshold is configurable via `BlockBasedTableOptions::uniform_cv_threshold` (default: 0.2).

unreleased_history/new_features/enforce_wbm_during_recovery.md

@@ -1 +0,0 @@
-Introduced enforce_write_buffer_manager_during_recovery option to allow WriteBufferManager to be enforced during WAL recovery. (Default: true)

unreleased_history/new_features/memtable_batch_lookup_optimization.md

@@ -1 +0,0 @@
-Add `memtable_batch_lookup_optimization` option to use batch lookup optimization for memtable MultiGet. For skip list memtables, after each key lookup, the search path is cached and reused for the next key, reducing per-key cost from O(log N) to O(log d) where d is the distance between consecutive keys. Benchmarks show ~7% improvement in memtable-resident MultiGet throughput.

unreleased_history/new_features/prepopulate_block_cache_on_compaction.md

@@ -1 +0,0 @@
-Added `BlockBasedTableOptions::PrepopulateBlockCache::kFlushAndCompaction` to prepopulate the block cache during both flush and compaction. Compaction-warmed blocks are inserted at `BOTTOM` priority (vs `LOW` for flush) so they are evicted first under cache pressure. Recommended only for use cases where most or all of the database is expected to reside in cache (e.g., tiered or remote storage where the working set fits in cache).

unreleased_history/new_features/verify_manifest_content_on_close.md

@@ -1 +0,0 @@
-Added new mutable DB option `verify_manifest_content_on_close` (default: false). When enabled, on DB close the MANIFEST file is read back and all records are validated (CRC checksums and logical content). If corruption is detected, a fresh MANIFEST is written from in-memory state.

util/io_dispatcher_imp.cc

@@ -168,6 +168,15 @@ Status ReadSet::ReadIndex(size_t block_index, CachableEntry<Block>* out) {
   // SubmitJob)
   if (pinned_blocks_[block_index].GetValue()) {
     *out = std::move(pinned_blocks_[block_index]);
+    // 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;
+    }
     // Note: Statistics for this block were already counted during SubmitJob
     // (either as cache hit or sync read)
     return Status::OK();
@@ -190,6 +199,14 @@ Status ReadSet::ReadIndex(size_t block_index, CachableEntry<Block>* out) {
       // After polling, the block should be in pinned_blocks_
       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;
+        }
         return Status::OK();
       }
 
@@ -197,8 +214,12 @@ Status ReadSet::ReadIndex(size_t block_index, CachableEntry<Block>* out) {
     }
   }
 
-  // Case 3: Block needs synchronous read
-  // If this block was pending prefetch, remove it since we're reading it now
+  // Case 3: Block needs synchronous read (pending or never-dispatched blocks).
+  // No ReleaseMemory() needed here because blocks reaching this path never had
+  // TryAcquireMemory() called — they were either pending prefetch or skipped
+  // during SubmitJob. block_sizes_[block_index] may be > 0 (set during
+  // SubmitJob for all uncached blocks) but that does not imply memory was
+  // acquired.
   RemoveFromPending(block_index);
 
   Status s = SyncRead(block_index);

util/io_dispatcher_test.cc

@@ -1791,6 +1791,125 @@ TEST_F(IODispatcherTest, CoalescedGroupsSplitByMemoryBudget) {
   }
 }
 
+// Regression tests for a bug where ReadIndex moved values out of
+// pinned_blocks_ via std::move, but neither ReleaseBlock() nor the destructor
+// released memory accounting because they checked pinned_blocks_.GetValue()
+// which was null after the move.
+// Tests run with both sync and async IO modes to cover Case 1 and Case 2
+// in ReadIndex().
+TEST_F(IODispatcherTest, MemoryReleasedAfterReadIndexThenReleaseBlock) {
+  for (bool async : {false, true}) {
+    // Skip async if io_uring not available
+    if (async && !kIOUringPresent) {
+      continue;
+    }
+    SCOPED_TRACE("async_io=" + std::to_string(async));
+
+    auto stats = CreateDBStatistics();
+    IODispatcherOptions opts;
+    opts.max_prefetch_memory_bytes = 100 * 1024;  // 100KB
+    opts.statistics = stats.get();
+    std::unique_ptr<IODispatcher> dispatcher(NewIODispatcher(opts));
+
+    std::unique_ptr<BlockBasedTable> table;
+    std::vector<BlockHandle> block_handles;
+    Status s = CreateAndOpenSST(20, &table, &block_handles);
+    ASSERT_OK(s);
+    ASSERT_GT(block_handles.size(), 0);
+
+    auto job = std::make_shared<IOJob>();
+    job->block_handles = block_handles;
+    job->table = table.get();
+    job->job_options.read_options.async_io = async;
+
+    std::shared_ptr<ReadSet> read_set;
+    s = dispatcher->SubmitJob(job, &read_set);
+    ASSERT_OK(s);
+    ASSERT_NE(read_set, nullptr);
+
+    // Some memory should have been granted for prefetch
+    ASSERT_GT(stats->getTickerCount(PREFETCH_MEMORY_BYTES_GRANTED), 0);
+
+    // Read all blocks — ReadIndex moves values out of pinned_blocks_.
+    // This also triggers TryDispatchPendingPrefetches as memory is released,
+    // which acquires more memory for pending groups. So granted grows during
+    // this loop.
+    for (size_t i = 0; i < block_handles.size(); ++i) {
+      CachableEntry<Block> block;
+      ASSERT_OK(read_set->ReadIndex(i, &block));
+      ASSERT_NE(block.GetValue(), nullptr);
+    }
+
+    // Release all blocks — should be a no-op for memory accounting since
+    // ReadIndex already released memory when moving values out
+    for (size_t i = 0; i < block_handles.size(); ++i) {
+      read_set->ReleaseBlock(i);
+    }
+
+    // Read both counters after all operations complete, since
+    // TryDispatchPendingPrefetches during ReadIndex may have granted additional
+    // memory for pending groups
+    uint64_t granted = stats->getTickerCount(PREFETCH_MEMORY_BYTES_GRANTED);
+    uint64_t released = stats->getTickerCount(PREFETCH_MEMORY_BYTES_RELEASED);
+    // With the bug, released < granted because ReleaseBlock skips
+    // ReleaseMemory when pinned_blocks_ value was already moved out
+    EXPECT_EQ(released, granted);
+  }
+}
+
+// Test that ReadSet destructor releases memory for blocks that were read
+// via ReadIndex but never explicitly released via ReleaseBlock.
+TEST_F(IODispatcherTest, DestructorReleasesMemoryAfterReadIndex) {
+  for (bool async : {false, true}) {
+    // Skip async if io_uring not available
+    if (async && !kIOUringPresent) {
+      continue;
+    }
+    SCOPED_TRACE("async_io=" + std::to_string(async));
+
+    auto stats = CreateDBStatistics();
+    IODispatcherOptions opts;
+    opts.max_prefetch_memory_bytes = 100 * 1024;  // 100KB
+    opts.statistics = stats.get();
+    std::unique_ptr<IODispatcher> dispatcher(NewIODispatcher(opts));
+
+    std::unique_ptr<BlockBasedTable> table;
+    std::vector<BlockHandle> block_handles;
+    Status s = CreateAndOpenSST(20, &table, &block_handles);
+    ASSERT_OK(s);
+    ASSERT_GT(block_handles.size(), 0);
+
+    {
+      auto job = std::make_shared<IOJob>();
+      job->block_handles = block_handles;
+      job->table = table.get();
+      job->job_options.read_options.async_io = async;
+
+      std::shared_ptr<ReadSet> read_set;
+      s = dispatcher->SubmitJob(job, &read_set);
+      ASSERT_OK(s);
+      ASSERT_NE(read_set, nullptr);
+
+      uint64_t granted = stats->getTickerCount(PREFETCH_MEMORY_BYTES_GRANTED);
+      ASSERT_GT(granted, 0);
+
+      // Read all blocks via ReadIndex (moves values out of pinned_blocks_)
+      // but do NOT call ReleaseBlock — let the destructor handle cleanup
+      for (size_t i = 0; i < block_handles.size(); ++i) {
+        CachableEntry<Block> block;
+        ASSERT_OK(read_set->ReadIndex(i, &block));
+      }
+      // read_set goes out of scope — destructor should release all memory
+    }
+
+    uint64_t granted = stats->getTickerCount(PREFETCH_MEMORY_BYTES_GRANTED);
+    uint64_t released = stats->getTickerCount(PREFETCH_MEMORY_BYTES_RELEASED);
+    // Destructor should release memory for all prefetched blocks,
+    // even those whose values were moved out by ReadIndex
+    EXPECT_EQ(released, granted);
+  }
+}
+
 }  // namespace ROCKSDB_NAMESPACE
 
 int main(int argc, char** argv) {