Update HISTORY.md for 11.13.2

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

Fix a RocksDB secondary with `open_files_async = true` latching on `PathNotFound` during `TryCatchUpWithPrimary()`.

**Why:** With `open_files_async = true` and `max_open_files = -1`, a secondary's catch-up could fail with `PathNotFound` on a table that was live at open time (most reachably in a non-default column family) but which the primary compacted away before the background async opener pinned it. The error latched in the tailer's status, so the secondary never advanced again and had to be closed and reopened. With `open_files_async = false` the same sequence caught up fine. This blocks enabling async secondary open where it would otherwise cut cold-open time substantially.

**Previous behavior:** In `VersionEditHandlerPointInTime::MaybeCreateVersionBeforeApplyEdit`, a negative-edge `LoadTableHandlers()` failure skipped the unrecoverable point-in-time only when the status was `Corruption`; a `PathNotFound` was returned as-is and latched. During a secondary's catch-up, `open_files_async` leaves recovery-time tables unpinned until the background opener reaches them, so `LoadTableHandlers()` could try to open a file the primary had already deleted.

**New behavior:** That negative-edge failure path now also treats `PathNotFound` as a missing file (skip this point-in-time and continue) when `no_error_if_files_missing_` is set, mirroring the tolerance `VersionEditHandler::LoadTables()` already applies. Catch-up continues past the transient state and installs the current version, so the secondary advances instead of latching.

**Implementation:**
- `db/version_edit_handler.cc`: broaden the negative-edge skip condition in `VersionEditHandlerPointInTime::MaybeCreateVersionBeforeApplyEdit` from `s.IsCorruption()` to also include `s.IsPathNotFound() && no_error_if_files_missing_`.
- `db/db_secondary_test.cc`: add `DBSecondaryTest.OpenFilesAsyncNonDefaultCfCatchUp`, a regression test that opens an async secondary with a non-default column family, holds the background opener back so the open-time table stays unpinned, has the primary flush and `CompactRange(kForce)` to replace and delete that table, then asserts `TryCatchUpWithPrimary()` succeeds and reads the new value.

**Safety boundaries:** The change only affects point-in-time handlers that opted into missing-file tolerance (`no_error_if_files_missing_`, set by secondary/follower tailing and best-efforts recovery); it does not relax error handling for strict recovery. It does not change the `trust_manifest_recovery_` / `defer_sst_file_opening_` paths, which already skip opening.

**Validation and limitations:** The regression test covers async secondary catch-up after deletion of an open-time SST in a non-default column family. Validation covers async opening, catch-up, memtable retention, and repeated scheduling changes. Strict recovery, API and file formats are unchanged. The full supported platform matrix has not been run locally.

Reviewed By: jaykorean

Differential Revision: D125378257

fbshipit-source-id: 3dd23c2f763e4e744c9f38fb77dfdfb3f554ee7f
(cherry picked from commit 81369fc170)

HISTORY.md

@@ -1,6 +1,51 @@
 # Rocksdb Change Log
 > NOTE: Entries for next release do not go here. Follow instructions in `unreleased_history/README.txt`
 
+## 11.13.2 (10/28/2026)
+### Bug Fixes
+* Fixed a secondary instance with `open_files_async = true` getting stuck after `TryCatchUpWithPrimary()` returned `PathNotFound` for a table file that the primary deleted before the background opener reached it.
+* Fixed blob direct write with more than one concurrent flush of a column family: a later flush sealed and published the blob files of the oldest pending memtable instead of its own, so flushes failed with `Corruption: VersionBuilder: Blob file #N already added` and the database stopped.
+
+## 11.13.1 (10/23/2026)
+### Bug Fixes
+* Fixed data loss with `min_tombstones_for_range_conversion` and external file ingestion. An iterator whose view predated the ingestion could convert the deleted keys around the ingested keys into a range tombstone that deleted them. This happened when the file got no new sequence number (sequence number zero, `ingest_behind`, or `allow_db_generated_files`), and when an iterator was created during the ingestion.
+* Fixed a data loss bug where an iterator with `ReadOptions::read_tier = kMemtableTier` could, with `min_tombstones_for_range_conversion` enabled, insert a range tombstone that deleted live keys stored only in SST files. Read-path range tombstone conversion now runs only for iterators with `read_tier = kReadAllTier`.
+
+## 11.13.0 (10/21/2026)
+### New Features
+* Added `FileSystem::IsPreallocationSupported()`. A `FileSystem` that never preallocates file space can return false so that DB open skips reopening and truncating the last WAL file.
+
+### Public API Changes
+* Added the experimental `DBOptions::track_options_file_number_in_manifest`
+rollout option (with C and Java bindings). The option is persisted in each
+OPTIONS file. When enabled, RocksDB durably creates an OPTIONS file and then
+commits its file number in MANIFEST. Column-family create/drop commits the
+configuration snapshot and membership in one MANIFEST record. A newer tracked
+OPTIONS file without a commit is ignored after a crash, while a newer OPTIONS
+file written with tracking disabled retains the legacy highest-numbered-file
+behavior. Batched column-family changes publish one snapshot; after partial
+success it can be a safe superset that persisted-options APIs filter against
+live MANIFEST membership. Ordinary `DB::Open` continues to use caller-supplied
+options and does not depend on the operational snapshot.
+* Added `EncodeEntry()`, `PackSequenceAndType()`, and `UnPackSequenceAndType()` in `rocksdb/utilities/types_util.h` to construct internal keys and pack or unpack their sequence numbers and entry types. `kInternalKeyFooterSize` exposes the footer size in bytes.
+* Added `ldb update_manifest --rebind_options_file --options_file=OPTIONS-<number>` for offline repair of a committed MANIFEST OPTIONS file number. Operators must select a trusted snapshot that has MANIFEST tracking enabled and provide any external DB/CF allocation roots that are not serialized in OPTIONS.
+
+### Behavior Changes
+* With `max_bgerror_resume_count <= 0` (auto resume disabled), a retryable or file-scoped IO error during compaction or flush is now a hard background error that stops writes until `DB::Resume()` is called. Previously, compaction errors were silently retried, and flush errors with a healthy WAL were soft errors that let writes continue while flushes were stopped.
+
+### Bug Fixes
+* Fixed two more `SstFileManager` trash accounting leaks: retrying a failed trash deletion immediately (for example with rate limiting disabled) after the file had disappeared, and a trash file that was deleted but whose directory sync then failed. In both cases the file's size stayed in the total trash size and tracked size until restart.
+* Fixed `SstFileManager` trash accounting leaking the size of a queued trash file that disappeared before deletion (for example because its directory was renamed). The leaked bytes stayed in the total trash size until restart, which could push every later deletion past `max_trash_db_ratio` and make it immediate instead of rate limited.
+* Fixed `MultiGet` with `ReadOptions::async_io` returning `NotSupported` when io_uring initialization or an asynchronous read submission fails. It now checks async I/O support on the reading thread and falls back to synchronous reads when needed.
+* Fixed subset checkpoints so their newest OPTIONS file lists only the column
+families retained by the checkpoint. This applies with both legacy
+highest-numbered OPTIONS selection and MANIFEST-tracked OPTIONS selection, and
+prevents `LoadLatestOptions` from recreating column families that the subset
+checkpoint intentionally dropped from its copied MANIFEST.
+
+### Performance Improvements
+* Perf context counters and timers no longer call the thread-local init function of `perf_context` on every update. With `kEnableCount`, a counter update costs about 60% less.
+
 ## 11.12.0 (10/07/2026)
 ### New Features
 * Added `FullExternalTableFactory` and related full-mode external table APIs that preserve RocksDB internal keys, enabling external tables to support nonzero sequence numbers, point and single deletions, merge operands, blob indexes, wide-column entities, snapshots, live DB writes, and multi-level compactions.

db/blob/blob_file_partition_manager.cc

@@ -543,10 +543,11 @@ Status BlobFilePartitionManager::WriteBlob(
   return Status::OK();
 }
 
-void BlobFilePartitionManager::RotateCurrentGeneration() {
+void BlobFilePartitionManager::RotateCurrentGeneration(uint64_t memtable_id) {
   MutexLock lock(&mutex_);
 
   GenerationBatch batch;
+  batch.memtable_id = memtable_id;
   batch.sealed_files = std::move(current_generation_sealed_files_);
   current_generation_sealed_files_.clear();
 
@@ -569,8 +570,19 @@ void BlobFilePartitionManager::RotateCurrentGeneration() {
   pending_generations_.emplace_back(std::move(batch));
 }
 
+BlobFilePartitionManager::GenerationBatch*
+BlobFilePartitionManager::FindPendingGeneration(uint64_t memtable_id) {
+  for (auto& batch : pending_generations_) {
+    if (batch.memtable_id == memtable_id) {
+      return &batch;
+    }
+  }
+  return nullptr;
+}
+
 Status BlobFilePartitionManager::PrepareFlushAdditions(
-    const WriteOptions& write_options, size_t num_generations,
+    const WriteOptions& write_options,
+    const std::vector<uint64_t>& memtable_ids,
     std::vector<BlobFileAddition>* additions,
     std::vector<BlobFileGarbage>* garbages,
     std::vector<std::vector<uint64_t>>* generation_blob_file_numbers) {
@@ -583,13 +595,13 @@ Status BlobFilePartitionManager::PrepareFlushAdditions(
   }
 
   MutexLock lock(&mutex_);
-  if (num_generations > pending_generations_.size()) {
-    return Status::Corruption(
-        "Missing blob direct write generation metadata for flush");
-  }
-
-  for (size_t i = 0; i < num_generations; ++i) {
-    GenerationBatch& batch = pending_generations_[i];
+  for (const uint64_t memtable_id : memtable_ids) {
+    GenerationBatch* const batch_ptr = FindPendingGeneration(memtable_id);
+    if (batch_ptr == nullptr) {
+      return Status::Corruption(
+          "Missing blob direct write generation metadata for flush");
+    }
+    GenerationBatch& batch = *batch_ptr;
     while (!batch.deferred_files.empty()) {
       DeferredFile deferred = std::move(batch.deferred_files.front());
       batch.deferred_files.pop_front();
@@ -668,12 +680,22 @@ Status BlobFilePartitionManager::MarkBlobWriteAsGarbage(uint64_t file_number,
   return Status::Corruption(message);
 }
 
-void BlobFilePartitionManager::CommitPreparedGenerations(
-    size_t num_generations) {
-  MutexLock lock(&mutex_);
-  while (num_generations-- > 0 && !pending_generations_.empty()) {
-    pending_generations_.pop_front();
+void BlobFilePartitionManager::CommitFlushedGenerations(
+    uint64_t earliest_unflushed_memtable_id) {
+  std::vector<uint64_t> sealed_file_numbers;
+  {
+    MutexLock lock(&mutex_);
+    while (!pending_generations_.empty() &&
+           pending_generations_.front().memtable_id <
+               earliest_unflushed_memtable_id) {
+      for (const auto& sealed_file :
+           pending_generations_.front().sealed_files) {
+        sealed_file_numbers.push_back(sealed_file.addition.GetBlobFileNumber());
+      }
+      pending_generations_.pop_front();
+    }
   }
+  RemoveFilePartitionMappings(sealed_file_numbers);
 }
 
 Status BlobFilePartitionManager::SyncAllOpenFiles(

db/blob/blob_file_partition_manager.h

@@ -88,27 +88,31 @@ class BlobFilePartitionManager {
                                      const Slice& key,
                                      const WideColumns& columns) const;
 
-  // Move the current active partition files into the next immutable
-  // memtable-generation batch. Called from SwitchMemtable() while DB mutex is
-  // held. No I/O is performed here.
-  void RotateCurrentGeneration();
+  // Move the current active partition files into the generation of the
+  // memtable `memtable_id`, which just became immutable. Called from
+  // SwitchMemtable() while DB mutex is held. No I/O is performed here.
+  void RotateCurrentGeneration(uint64_t memtable_id);
 
-  // Seal the first `num_generations` queued immutable generations and return
-  // all blob additions and initial-garbage updates that must be registered
-  // with the matching flush. Generations stay queued until
-  // CommitPreparedGenerations() is called. If `generation_blob_file_numbers`
-  // is non-null, it receives the sealed blob file numbers for each prepared
-  // generation in the same FIFO order.
+  // Seal the queued generations of the memtables `memtable_ids` and return all
+  // blob additions and initial-garbage updates that must be registered with
+  // the flush of those memtables. Concurrent flushes pick different
+  // memtables, so generations are matched by memtable ID, not by queue
+  // position. Generations stay queued until CommitFlushedGenerations() removes
+  // them. If `generation_blob_file_numbers` is non-null, it receives the
+  // sealed blob file numbers for each memtable, in `memtable_ids` order.
   Status PrepareFlushAdditions(const WriteOptions& write_options,
-                               size_t num_generations,
+                               const std::vector<uint64_t>& memtable_ids,
                                std::vector<BlobFileAddition>* additions,
                                std::vector<BlobFileGarbage>* garbages,
                                std::vector<std::vector<uint64_t>>*
                                    generation_blob_file_numbers = nullptr);
 
-  // Remove the first `num_generations` prepared immutable generations after
-  // their blob-file additions and garbage edits were committed to MANIFEST.
-  void CommitPreparedGenerations(size_t num_generations);
+  // Remove the generations of the memtables older than
+  // `earliest_unflushed_memtable_id`, whose flush results are installed, and
+  // stop tracking their sealed files as manager-owned. A flush that finishes
+  // before an older one is installed with it, or rolled back if it fails, so
+  // the flush that prepared a generation cannot remove it on its own.
+  void CommitFlushedGenerations(uint64_t earliest_unflushed_memtable_id);
 
   // In v1 flush-on-write mode each AddRecord flushes to the OS immediately, so
   // there is nothing extra to do for the non-sync write path.
@@ -200,6 +204,8 @@ class BlobFilePartitionManager {
   };
 
   struct GenerationBatch {
+    // The memtable that became immutable when this generation was rotated.
+    uint64_t memtable_id = 0;
     // Files that were still open when the memtable became immutable.
     std::deque<DeferredFile> deferred_files;
     // Files already sealed for this memtable generation.
@@ -208,6 +214,8 @@ class BlobFilePartitionManager {
 
   // Clears all active-file bookkeeping from a partition after sealing.
   void ResetPartitionState(Partition* partition);
+  // Returns the queued generation of `memtable_id`, or null.
+  GenerationBatch* FindPendingGeneration(uint64_t memtable_id);
   // Marks a blob file as manager-owned until it is committed or abandoned.
   void AddFilePartitionMapping(uint64_t file_number, uint32_t partition_idx);
   // Stops tracking a blob file after open/seal failure.

db/blob/db_blob_direct_write_test.cc

@@ -10,7 +10,9 @@
 
 #include <algorithm>
 #include <array>
+#include <atomic>
 #include <functional>
+#include <future>
 #include <limits>
 #include <mutex>
 #include <set>
@@ -1460,7 +1462,9 @@ TEST_F(DBBlobDirectWriteTest,
   std::vector<BlobFileGarbage> garbages1;
   // Simulate a flush retry: prepare once, skip commit, then prepare the same
   // generation again and verify it reuses the already sealed blob file.
-  ASSERT_OK(mgr->PrepareFlushAdditions(WriteOptions(), /*num_generations=*/1,
+  const std::vector<uint64_t> memtable_ids = {
+      cfh->cfd()->imm()->GetEarliestMemTableID()};
+  ASSERT_OK(mgr->PrepareFlushAdditions(WriteOptions(), memtable_ids,
                                        &additions1, &garbages1));
   ASSERT_EQ(additions1.size(), 1U);
   ASSERT_TRUE(garbages1.empty());
@@ -1469,7 +1473,7 @@ TEST_F(DBBlobDirectWriteTest,
 
   std::vector<BlobFileAddition> additions2;
   std::vector<BlobFileGarbage> garbages2;
-  ASSERT_OK(mgr->PrepareFlushAdditions(WriteOptions(), /*num_generations=*/1,
+  ASSERT_OK(mgr->PrepareFlushAdditions(WriteOptions(), memtable_ids,
                                        &additions2, &garbages2));
   ASSERT_EQ(additions2.size(), 1U);
   ASSERT_TRUE(garbages2.empty());
@@ -1875,6 +1879,131 @@ TEST_F(DBBlobDirectWriteTest,
   AssertOrderedTraceStoresLogicalPut(options, /*expect_blob_files=*/false);
 }
 
+TEST_F(DBBlobDirectWriteTest, OverlappingFlushesPublishTheirOwnBlobFiles) {
+  for (bool atomic_flush : {false, true}) {
+    SCOPED_TRACE("atomic_flush=" + std::to_string(atomic_flush));
+    Options options = GetDirectWriteOptions();
+    options.atomic_flush = atomic_flush;
+    options.max_background_jobs = 8;
+    options.max_write_buffer_number = 4;
+    options.disable_auto_compactions = true;
+    DestroyAndReopen(options);
+    const std::string value(1024, 'v');
+
+    // The flush of the memtable holding "key1" waits while it writes its
+    // table file, until a second flush, of the memtable holding "key2", has
+    // finished its own table file.
+    std::atomic<int> table_writes{0};
+    std::atomic<bool> released{false};
+    std::promise<void> first_flush_writing;
+    std::promise<void> release_first_flush;
+    std::shared_future<void> release_first_flush_future =
+        release_first_flush.get_future().share();
+    SyncPoint::GetInstance()->SetCallBack(
+        "FlushJob::WriteLevel0Table:num_memtables", [&](void*) {
+          if (table_writes.fetch_add(1) == 0) {
+            first_flush_writing.set_value();
+            release_first_flush_future.wait();
+          }
+        });
+    SyncPoint::GetInstance()->SetCallBack(
+        atomic_flush ? "DBImpl::AtomicFlushMemTablesToOutputFiles:WaitCV"
+                     : "DBImpl::FlushMemTableToOutputFile:Finish",
+        [&](void*) {
+          if (!released.exchange(true)) {
+            release_first_flush.set_value();
+          }
+        });
+    SyncPoint::GetInstance()->EnableProcessing();
+
+    FlushOptions no_wait;
+    no_wait.wait = false;
+    no_wait.allow_write_stall = true;
+    ASSERT_OK(Put("key1", value));
+    ASSERT_OK(db_->Flush(no_wait));
+    first_flush_writing.get_future().wait();
+    ASSERT_OK(Put("key2", value));
+    ASSERT_OK(db_->Flush(no_wait));
+
+    ASSERT_OK(dbfull()->TEST_WaitForFlushMemTable());
+    SyncPoint::GetInstance()->DisableProcessing();
+    SyncPoint::GetInstance()->ClearAllCallBacks();
+    ASSERT_OK(dbfull()->TEST_GetBGError());
+    ASSERT_EQ(2, NumTableFilesAtLevel(0));
+
+    Reopen(options);
+    ASSERT_EQ(value, Get("key1"));
+    ASSERT_EQ(value, Get("key2"));
+  }
+}
+
+TEST_F(DBBlobDirectWriteTest, RetryAfterOlderOverlappingFlushFails) {
+  Options options = GetDirectWriteOptions();
+  options.max_background_jobs = 8;
+  options.max_write_buffer_number = 4;
+  options.disable_auto_compactions = true;
+  options.max_bgerror_resume_count = 0;
+  DestroyAndReopen(options);
+  const std::string value(1024, 'v');
+
+  // The flush of the memtable holding "key2" finishes while the older flush,
+  // of the memtable holding "key1", is still writing its table file. The
+  // newer flush's results wait to be installed after the older one's, but the
+  // older flush fails, which rolls the newer memtable back too.
+  std::atomic<int> table_writes{0};
+  std::atomic<bool> released{false};
+  std::atomic<bool> failed{false};
+  std::promise<void> first_flush_writing;
+  std::promise<void> release_first_flush;
+  std::shared_future<void> release_first_flush_future =
+      release_first_flush.get_future().share();
+  SyncPoint::GetInstance()->SetCallBack(
+      "FlushJob::WriteLevel0Table:num_memtables", [&](void*) {
+        if (table_writes.fetch_add(1) == 0) {
+          first_flush_writing.set_value();
+          release_first_flush_future.wait();
+        }
+      });
+  SyncPoint::GetInstance()->SetCallBack(
+      "DBImpl::FlushMemTableToOutputFile:Finish", [&](void*) {
+        if (!released.exchange(true)) {
+          release_first_flush.set_value();
+        }
+      });
+  SyncPoint::GetInstance()->SetCallBack(
+      "FlushJob::WriteLevel0Table:s", [&](void* arg) {
+        if (released.load() && !failed.exchange(true)) {
+          IOStatus injected = IOStatus::IOError("injected");
+          injected.SetRetryable(true);
+          *static_cast<Status*>(arg) = injected;
+        }
+      });
+  SyncPoint::GetInstance()->EnableProcessing();
+
+  FlushOptions no_wait;
+  no_wait.wait = false;
+  no_wait.allow_write_stall = true;
+  ASSERT_OK(Put("key1", value));
+  ASSERT_OK(db_->Flush(no_wait));
+  first_flush_writing.get_future().wait();
+  ASSERT_OK(Put("key2", value));
+  ASSERT_OK(db_->Flush(no_wait));
+
+  ASSERT_NOK(dbfull()->TEST_WaitForFlushMemTable());
+  SyncPoint::GetInstance()->DisableProcessing();
+  SyncPoint::GetInstance()->ClearAllCallBacks();
+  ASSERT_TRUE(failed.load());
+
+  // Retrying the flush must still find the blob files of both memtables.
+  ASSERT_OK(dbfull()->Resume());
+  ASSERT_OK(dbfull()->TEST_WaitForFlushMemTable());
+  ASSERT_OK(dbfull()->TEST_GetBGError());
+
+  Reopen(options);
+  ASSERT_EQ(value, Get("key1"));
+  ASSERT_EQ(value, Get("key2"));
+}
+
 class DBBlobDirectWriteWithTimestampTest : public DBBasicTestWithTimestampBase {
  protected:
   DBBlobDirectWriteWithTimestampTest()

db/db_flush_test.cc

@@ -4433,7 +4433,8 @@ TEST_F(DBFlushTest, FlushAfterReadPathRangeTombstoneInsertion) {
         // already immutable. Without the fix, this succeeds and creates
         // an entry count mismatch.
         cfh->cfd()->mem()->AddLogicallyRedundantRangeTombstone(
-            1 /* seq */, "b", "f", cfh->cfd()->GetIngestSstLock());
+            1 /* seq */, "b", "f", 0 /* version_number */,
+            cfh->cfd()->GetIngestSstLock());
       });
   SyncPoint::GetInstance()->EnableProcessing();
 

db/db_impl/db_impl.cc

@@ -7861,6 +7861,20 @@ Status DBImpl::CommitFileIngestionHandles(
           versions_->SetLastSequence(max_assigned_seqno);
         }
       }
+      // An iterator whose Version predates the ingestion cannot see the
+      // ingested files, whatever sequence number it reads at: files can get
+      // no new seqno (seqno zero, ingest_behind, DB-generated files), and an
+      // iterator can pair an older SuperVersion with a newer sequence number.
+      // Such iterators must not convert tombstones over the ingested files.
+      // We still hold the per-CF ingest_sst_lock as a ReadLock, so no
+      // conversion runs until the barrier is in place.
+      if (status.ok()) {
+        for (auto* job : ingestion_jobs) {
+          auto* cfd = job->GetColumnFamilyData();
+          cfd->mem()->BumpIngestVersionBarrier(
+              cfd->current()->GetVersionNumber());
+        }
+      }
     }
 
     for (auto* job : ingestion_jobs) {

db/db_impl/db_impl_compaction_flush.cc

@@ -62,6 +62,15 @@ void RecordAtomicFlushRequestReason(Statistics* stats,
   }
 }
 
+std::vector<uint64_t> MemTableIds(const autovector<ReadOnlyMemTable*>& mems) {
+  std::vector<uint64_t> ids;
+  ids.reserve(mems.size());
+  for (const ReadOnlyMemTable* mem : mems) {
+    ids.push_back(mem->GetID());
+  }
+  return ids;
+}
+
 }  // namespace
 
 bool DBImpl::EnoughRoomForCompaction(
@@ -311,8 +320,7 @@ Status DBImpl::FlushMemTableToOutputFile(
                      flush_reason);
 
   bool switched_to_mempurge = false;
-  size_t prepared_blob_generations = 0;
-  std::vector<uint64_t> sealed_blob_numbers;
+  std::vector<uint64_t> prepared_blob_memtable_ids;
   // Within flush_job.Run, rocksdb may call event listener to notify
   // file creation and deletion.
   //
@@ -323,17 +331,17 @@ Status DBImpl::FlushMemTableToOutputFile(
     auto mgr_handle = cfd->blob_partition_manager_handle();
     auto* mgr = mgr_handle.get();
     if (mgr != nullptr) {
-      prepared_blob_generations = flush_job.GetMemTables().size();
+      prepared_blob_memtable_ids = MemTableIds(flush_job.GetMemTables());
       std::vector<BlobFileAddition> write_path_additions;
       std::vector<BlobFileGarbage> write_path_garbages;
       std::vector<std::vector<uint64_t>> generation_blob_file_numbers;
       mutex_.Unlock();
       s = mgr->PrepareFlushAdditions(
-          write_options, prepared_blob_generations, &write_path_additions,
+          write_options, prepared_blob_memtable_ids, &write_path_additions,
           &write_path_garbages, &generation_blob_file_numbers);
       mutex_.Lock();
       if (!s.ok()) {
-        prepared_blob_generations = 0;
+        prepared_blob_memtable_ids.clear();
       } else if (!write_path_additions.empty()) {
         const auto& memtables = flush_job.GetMemTables();
         assert(generation_blob_file_numbers.size() == memtables.size());
@@ -344,9 +352,6 @@ Status DBImpl::FlushMemTableToOutputFile(
           memtables[i]->ProtectSealedBlobFiles(mgr_handle,
                                                generation_blob_file_numbers[i]);
         }
-        for (const auto& addition : write_path_additions) {
-          sealed_blob_numbers.push_back(addition.GetBlobFileNumber());
-        }
         flush_job.AddExternalBlobFileAdditions(std::move(write_path_additions));
         flush_job.AddExternalBlobFileGarbages(std::move(write_path_garbages));
       } else if (!write_path_garbages.empty()) {
@@ -390,7 +395,7 @@ Status DBImpl::FlushMemTableToOutputFile(
   });
 
   if (cfd->blob_partition_manager() != nullptr &&
-      prepared_blob_generations > 0) {
+      !prepared_blob_memtable_ids.empty()) {
     auto unconsumed_additions = flush_job.TakeExternalBlobFileAdditions();
     auto unconsumed_garbages = flush_job.TakeExternalBlobFileGarbages();
     if (!s.ok() || !unconsumed_additions.empty() ||
@@ -399,8 +404,7 @@ Status DBImpl::FlushMemTableToOutputFile(
       // in pending_generations_. The same immutable memtables still reference
       // these sealed blob files, so a later flush retry must publish the exact
       // same additions rather than abandoning them here.
-      prepared_blob_generations = 0;
-      sealed_blob_numbers.clear();
+      prepared_blob_memtable_ids.clear();
     }
   }
 
@@ -408,12 +412,9 @@ Status DBImpl::FlushMemTableToOutputFile(
     flush_job.Cancel();
   }
 
-  if (s.ok() && prepared_blob_generations > 0) {
-    auto* mgr = cfd->blob_partition_manager();
-    mgr->CommitPreparedGenerations(prepared_blob_generations);
-    if (!sealed_blob_numbers.empty()) {
-      mgr->RemoveFilePartitionMappings(sealed_blob_numbers);
-    }
+  if (s.ok() && !prepared_blob_memtable_ids.empty()) {
+    cfd->blob_partition_manager()->CommitFlushedGenerations(
+        cfd->imm()->GetEarliestMemTableID());
   }
 
   if (s.ok()) {
@@ -656,13 +657,12 @@ Status DBImpl::AtomicFlushMemTablesToOutputFiles(
   // <bool /* executed */, Status /* status code */>
   autovector<std::pair<bool, Status>> exec_status;
   std::vector<bool> pick_status;
-  std::vector<size_t> prepared_blob_generations(num_cfs, 0);
+  std::vector<std::vector<uint64_t>> prepared_blob_memtable_ids(num_cfs);
   for (int i = 0; i != num_cfs; ++i) {
     // Initially all jobs are not executed, with status OK.
     exec_status.emplace_back(false, Status::OK());
     pick_status.push_back(false);
   }
-  std::unordered_map<int, std::vector<uint64_t>> sealed_blob_numbers_by_cf;
 
   bool flush_for_recovery =
       bg_flush_args[0].flush_reason_ == FlushReason::kErrorRecovery ||
@@ -694,17 +694,17 @@ Status DBImpl::AtomicFlushMemTablesToOutputFiles(
         continue;
       }
 
-      prepared_blob_generations[i] = jobs[i]->GetMemTables().size();
+      prepared_blob_memtable_ids[i] = MemTableIds(jobs[i]->GetMemTables());
       std::vector<BlobFileAddition> write_path_additions;
       std::vector<BlobFileGarbage> write_path_garbages;
       std::vector<std::vector<uint64_t>> generation_blob_file_numbers;
       mutex_.Unlock();
       s = mgr->PrepareFlushAdditions(
-          write_options, prepared_blob_generations[i], &write_path_additions,
+          write_options, prepared_blob_memtable_ids[i], &write_path_additions,
           &write_path_garbages, &generation_blob_file_numbers);
       mutex_.Lock();
       if (!s.ok()) {
-        prepared_blob_generations[i] = 0;
+        prepared_blob_memtable_ids[i].clear();
         break;
       }
       if (!write_path_additions.empty()) {
@@ -714,10 +714,6 @@ Status DBImpl::AtomicFlushMemTablesToOutputFiles(
           memtables[j]->ProtectSealedBlobFiles(mgr_handle,
                                                generation_blob_file_numbers[j]);
         }
-        auto& sealed_blob_numbers = sealed_blob_numbers_by_cf[i];
-        for (const auto& addition : write_path_additions) {
-          sealed_blob_numbers.push_back(addition.GetBlobFileNumber());
-        }
         jobs[i]->AddExternalBlobFileAdditions(std::move(write_path_additions));
       }
       if (!write_path_garbages.empty()) {
@@ -987,7 +983,7 @@ Status DBImpl::AtomicFlushMemTablesToOutputFiles(
 
   for (int i = 0; i != num_cfs; ++i) {
     auto* mgr = cfds[i]->blob_partition_manager();
-    if (mgr == nullptr || prepared_blob_generations[i] == 0) {
+    if (mgr == nullptr || prepared_blob_memtable_ids[i].empty()) {
       continue;
     }
 
@@ -998,15 +994,11 @@ Status DBImpl::AtomicFlushMemTablesToOutputFiles(
       // Same retry semantics as the single-CF flush path above: do not
       // consume these prepared generations yet, because the rolled-back
       // memtables still need to reuse the same sealed blob files on retry.
-      prepared_blob_generations[i] = 0;
+      prepared_blob_memtable_ids[i].clear();
       continue;
     }
 
-    mgr->CommitPreparedGenerations(prepared_blob_generations[i]);
-    auto it = sealed_blob_numbers_by_cf.find(i);
-    if (it != sealed_blob_numbers_by_cf.end() && !it->second.empty()) {
-      mgr->RemoveFilePartitionMappings(it->second);
-    }
+    mgr->CommitFlushedGenerations(cfds[i]->imm()->GetEarliestMemTableID());
   }
 
   if (s.ok()) {

db/db_impl/db_impl_write.cc

@@ -3567,7 +3567,7 @@ Status DBImpl::SwitchMemtableImpl(ColumnFamilyData* cfd, WriteContext* context,
   }
 
   if (cfd->blob_partition_manager() != nullptr) {
-    cfd->blob_partition_manager()->RotateCurrentGeneration();
+    cfd->blob_partition_manager()->RotateCurrentGeneration(cfd->mem()->GetID());
   }
 
   cfd->mem()->SetNextLogNumber(cur_wal_number_);

db/db_iter.cc

@@ -96,6 +96,7 @@ DBIter::DBIter(Env* _env, const ReadOptions& read_options,
       trace_cf_id_(cfd != nullptr ? cfd->GetID() : 0),
       has_trace_state_(db_impl != nullptr && cfd != nullptr),
       ingest_sst_lock_(cfd != nullptr ? &cfd->GetIngestSstLock() : nullptr),
+      version_number_(version != nullptr ? version->GetVersionNumber() : 0),
       timestamp_ub_(read_options.timestamp),
       timestamp_lb_(read_options.iter_start_ts),
       timestamp_size_(timestamp_ub_ ? timestamp_ub_->size() : 0),
@@ -117,13 +118,15 @@ DBIter::DBIter(Env* _env, const ReadOptions& read_options,
                                      read_options.total_order_seek ||
                                      read_options.auto_prefix_mode),
       // Read-path range conversion assumes the scan can observe all interior
-      // live keys. table_filter can hide whole SSTs, and timestamp filtering
+      // live keys. table_filter can hide whole SSTs, read tiers other than
+      // kReadAllTier can skip SSTs or uncached blocks, and timestamp filtering
       // can hide newer UDT versions unless the read is at max timestamp with no
       // lower timestamp bound. Legacy prefix iterators without
       // prefix_same_as_start do not guarantee complete scans, so conversion
       // must stay disabled for the iterator lifetime.
       min_tombstones_for_range_conversion_(
           active_mem != nullptr && !HasTableFilter(read_options) &&
+                  read_options.read_tier == kReadAllTier &&
                   (expect_total_order_inner_iter_ || prefix_same_as_start_) &&
                   HasFullTimestampVisibility(read_options)
               ? mutable_cf_options.min_tombstones_for_range_conversion
@@ -1831,7 +1834,7 @@ void DBIter::MaybeInsertRangeTombstone(const Slice& end_key) {
   assert(ingest_sst_lock_ != nullptr);
   if (active_mem_->AddLogicallyRedundantRangeTombstone(
           insert_seq, range_tomb_first_key_.GetUserKey(), end_key,
-          *ingest_sst_lock_)) {
+          version_number_, *ingest_sst_lock_)) {
     RecordTick(statistics_, READ_PATH_RANGE_TOMBSTONES_INSERTED);
     ROCKS_LOG_DEBUG(logger_,
                     "Inserted range tombstone [%s, %s) @ seq %" PRIu64

db/db_iter.h

@@ -713,6 +713,8 @@ class DBIter final : public Iterator {
   uint32_t trace_cf_id_;
   bool has_trace_state_;
   port::RWMutex* ingest_sst_lock_;
+  // Number of the Version this iterator reads, or 0 if unknown.
+  const uint64_t version_number_;
   const Slice* const timestamp_ub_;
   const Slice* const timestamp_lb_;
   const size_t timestamp_size_;

db/db_iterator_test.cc

@@ -7555,8 +7555,9 @@ TEST_P(ReadPathRangeTombstoneTest, NoInsertionOnBlockCacheTierIncomplete) {
     ASSERT_TRUE(iter->status().IsIncomplete());
   }
 
-  // No range tombstone should have been inserted despite meeting threshold,
-  // because the iterator terminated with Incomplete (cache miss).
+  // No range tombstone should have been inserted despite meeting threshold:
+  // conversion is disabled for kBlockCacheTier, and the iterator terminated
+  // with Incomplete (cache miss).
   ASSERT_EQ(attempted_insert_ranges_.size(), 0);
   ASSERT_EQ(
       options.statistics->getTickerCount(READ_PATH_RANGE_TOMBSTONES_INSERTED),
@@ -7570,6 +7571,154 @@ TEST_P(ReadPathRangeTombstoneTest, NoInsertionOnBlockCacheTierIncomplete) {
   }
 }
 
+TEST_P(ReadPathRangeTombstoneTest, NoInsertionFromMemtableTierIterator) {
+  Options options = CurrentOptions();
+  options.min_tombstones_for_range_conversion = 2;
+  options.statistics = CreateDBStatistics();
+  DestroyAndReopen(options);
+
+  // "c" is live and only exists in an SST file, which a kMemtableTier
+  // iterator does not read.
+  ASSERT_OK(Put("c", "vc"));
+  ASSERT_OK(Flush());
+
+  ASSERT_OK(Put("a", "va"));
+  ASSERT_OK(Delete("b"));
+  ASSERT_OK(Delete("d"));
+  ASSERT_OK(Put("e", "ve"));
+
+  {
+    ReadOptions ro;
+    ro.read_tier = kMemtableTier;
+    std::unique_ptr<Iterator> iter(db_->NewIterator(ro));
+    std::vector<std::string> seen;
+    if (Forward()) {
+      for (iter->SeekToFirst(); iter->Valid(); iter->Next()) {
+        seen.push_back(iter->key().ToString());
+      }
+    } else {
+      for (iter->SeekToLast(); iter->Valid(); iter->Prev()) {
+        seen.insert(seen.begin(), iter->key().ToString());
+      }
+    }
+    ASSERT_OK(iter->status());
+    ASSERT_EQ(seen, std::vector<std::string>({"a", "e"}));
+  }
+
+  ASSERT_EQ(Get("c"), "vc");
+  ASSERT_EQ(attempted_insert_ranges_.size(), 0);
+  ASSERT_EQ(
+      options.statistics->getTickerCount(READ_PATH_RANGE_TOMBSTONES_INSERTED),
+      0);
+  ASSERT_OK(Flush());
+  ASSERT_OK(db_->CompactRange(CompactRangeOptions(), nullptr, nullptr));
+  ASSERT_EQ(Get("c"), "vc");
+}
+
+TEST_P(ReadPathRangeTombstoneTest, PartialTierReadsAfterConversion) {
+  Options options = CurrentOptions();
+  options.statistics = CreateDBStatistics();
+  BlockBasedTableOptions table_options;
+  table_options.flush_block_policy_factory =
+      std::make_shared<FlushBlockEveryKeyPolicyFactory>();
+  std::shared_ptr<Cache> block_cache = NewLRUCache(1 << 20);
+  table_options.block_cache = block_cache;
+  options.table_factory.reset(NewBlockBasedTableFactory(table_options));
+  // Conversion starts disabled so the reads recorded before it do not insert.
+  DestroyAndReopen(options);
+
+  for (char c = 'a'; c <= 'h'; c++) {
+    ASSERT_OK(Put(std::string(1, c), std::string("v") + c));
+  }
+  ASSERT_OK(Flush());
+  ASSERT_OK(Delete("b"));
+  ASSERT_OK(Delete("c"));
+  ASSERT_OK(Flush());
+  ASSERT_OK(Put("a", "va2"));
+  ASSERT_OK(Delete("d"));
+  ASSERT_OK(Put("e", "ve2"));
+  ASSERT_OK(Delete("e"));
+
+  auto scan = [&](ReadTier tier) {
+    ReadOptions ro;
+    ro.read_tier = tier;
+    std::unique_ptr<Iterator> iter(db_->NewIterator(ro));
+    std::vector<std::string> out;
+    if (Forward()) {
+      for (iter->SeekToFirst(); iter->Valid(); iter->Next()) {
+        out.push_back(iter->key().ToString() + "=" + iter->value().ToString());
+      }
+    } else {
+      for (iter->SeekToLast(); iter->Valid(); iter->Prev()) {
+        out.insert(out.begin(),
+                   iter->key().ToString() + "=" + iter->value().ToString());
+      }
+    }
+    out.push_back(iter->status().ToString());
+    return out;
+  };
+  // kMemtableTier only affects iterators, so point reads compare
+  // kBlockCacheTier with kReadAllTier.
+  auto gets = [&](ReadTier tier) {
+    ReadOptions ro;
+    ro.read_tier = tier;
+    std::vector<std::string> out;
+    for (char c = 'a'; c <= 'h'; c++) {
+      std::string value;
+      Status s = db_->Get(ro, std::string(1, c), &value);
+      out.push_back(std::string(1, c) + ":" + (s.ok() ? value : s.ToString()));
+    }
+    return out;
+  };
+
+  const auto cold_cache_scan_before = scan(kBlockCacheTier);
+  const auto cold_cache_gets_before = gets(kBlockCacheTier);
+  const std::string incomplete =
+      Status::Incomplete("no blocking io").ToString();
+  ASSERT_EQ(cold_cache_scan_before.back(), incomplete);
+  ASSERT_EQ(cold_cache_gets_before[1], "b:" + incomplete);
+  ASSERT_EQ(cold_cache_gets_before[2], "c:" + incomplete);
+
+  // A full scan with conversion disabled reads every block into the cache.
+  const std::vector<std::string> full_view = {"a=va2", "f=vf", "g=vg", "h=vh",
+                                              "OK"};
+  ASSERT_EQ(scan(kReadAllTier), full_view);
+  ASSERT_EQ(attempted_insert_ranges_.size(), 0);
+
+  const auto mem_scan_before = scan(kMemtableTier);
+  ASSERT_EQ(mem_scan_before, std::vector<std::string>({"a=va2", "OK"}));
+  const auto warm_cache_scan_before = scan(kBlockCacheTier);
+  ASSERT_EQ(warm_cache_scan_before, full_view);
+  const auto warm_cache_gets_before = gets(kBlockCacheTier);
+  const auto full_gets = gets(kReadAllTier);
+  ASSERT_EQ(warm_cache_gets_before, full_gets);
+
+  ASSERT_OK(db_->SetOptions({{"min_tombstones_for_range_conversion", "2"}}));
+  // A complete kBlockCacheTier scan does not convert. Only kReadAllTier does.
+  ASSERT_EQ(scan(kBlockCacheTier), warm_cache_scan_before);
+  ASSERT_EQ(attempted_insert_ranges_.size(), 0);
+  ASSERT_EQ(scan(kReadAllTier), full_view);
+  ASSERT_EQ(attempted_insert_ranges_.size(), 1);
+  AssertRange(0, "b", "f");
+  ASSERT_EQ(
+      options.statistics->getTickerCount(READ_PATH_RANGE_TOMBSTONES_INSERTED),
+      1);
+
+  ASSERT_EQ(scan(kMemtableTier), mem_scan_before);
+  ASSERT_EQ(scan(kBlockCacheTier), warm_cache_scan_before);
+  ASSERT_EQ(gets(kBlockCacheTier), warm_cache_gets_before);
+
+  // With a cold cache, Gets of the covered keys "b" and "c" now stop at the
+  // memtable range tombstone and return the full-tier NotFound instead of
+  // Incomplete. No read returns a different value.
+  block_cache->EraseUnRefEntries();
+  ASSERT_EQ(scan(kBlockCacheTier), cold_cache_scan_before);
+  auto expected_cold_cache_gets = cold_cache_gets_before;
+  expected_cold_cache_gets[1] = full_gets[1];
+  expected_cold_cache_gets[2] = full_gets[2];
+  ASSERT_EQ(gets(kBlockCacheTier), expected_cold_cache_gets);
+}
+
 TEST_P(ReadPathRangeTombstoneTest, SkipInsertionWhenCoveredByExistingRange) {
   Options options = CurrentOptions();
   options.min_tombstones_for_range_conversion = 4;
@@ -8265,6 +8414,361 @@ TEST_P(ReadPathRangeTombstoneTest,
   db_->ReleaseSnapshot(snap);
 }
 
+// An ingested file that gets no new sequence number is invisible to an
+// iterator created before the ingestion. That iterator must not convert the
+// deleted keys around the file into a range tombstone that covers it.
+TEST_P(ReadPathRangeTombstoneTest, IngestedFileSurvivesOlderIterator) {
+  enum class IngestMode { kSeqnoZero, kIngestBehind, kDbGenerated };
+  for (IngestMode mode : {IngestMode::kSeqnoZero, IngestMode::kIngestBehind,
+                          IngestMode::kDbGenerated}) {
+    SCOPED_TRACE("ingest mode " + std::to_string(static_cast<int>(mode)));
+    Options options = CurrentOptions();
+    options.min_tombstones_for_range_conversion = 2;
+    options.disable_auto_compactions = true;
+    options.cf_allow_ingest_behind = mode == IngestMode::kIngestBehind;
+    DestroyAndReopen(options);
+
+    std::string file = dbname_ + "/ingest_c.sst";
+    SequenceNumber expected_file_seqno = 0;
+    if (mode == IngestMode::kDbGenerated) {
+      const std::string src_path = dbname_ + "_src";
+      ASSERT_OK(DestroyDB(src_path, options));
+      std::unique_ptr<DB> src;
+      ASSERT_OK(DB::Open(options, src_path, &src));
+      ASSERT_OK(src->Put(WriteOptions(), "c", "vc"));
+      ASSERT_OK(src->Flush(FlushOptions()));
+      std::vector<LiveFileMetaData> src_files;
+      src->GetLiveFilesMetaData(&src_files);
+      ASSERT_EQ(src_files.size(), 1);
+      expected_file_seqno = src_files[0].largest_seqno;
+      ASSERT_GT(expected_file_seqno, 0);
+      file = src_files[0].directory + "/" + src_files[0].relative_filename;
+      ASSERT_OK(src->Close());
+    } else {
+      SstFileWriter writer(EnvOptions(), options);
+      ASSERT_OK(writer.Open(file));
+      ASSERT_OK(writer.Put("c", "vc"));
+      ASSERT_OK(writer.Finish());
+    }
+
+    for (const char* key : {"a", "b", "d", "e", "f", "g"}) {
+      ASSERT_OK(Put(key, std::string("v") + key));
+    }
+    ASSERT_OK(Flush());
+    ASSERT_OK(Delete("b"));
+    ASSERT_OK(Delete("d"));
+    // The iterator sequence number must be above the DB-generated file's.
+    ASSERT_OK(Delete("f"));
+    ASSERT_GT(db_->GetLatestSequenceNumber(), expected_file_seqno);
+
+    std::unique_ptr<Iterator> iter(db_->NewIterator(ReadOptions()));
+
+    // "c" falls between the deleted keys and overlaps no key in the DB.
+    IngestExternalFileOptions ifo;
+    ifo.ingest_behind = mode == IngestMode::kIngestBehind;
+    ifo.allow_db_generated_files = mode == IngestMode::kDbGenerated;
+    ASSERT_OK(db_->IngestExternalFile({file}, ifo));
+    std::vector<LiveFileMetaData> files;
+    db_->GetLiveFilesMetaData(&files);
+    bool found = false;
+    for (const auto& f : files) {
+      if (f.smallestkey == "c") {
+        found = true;
+        ASSERT_EQ(f.level, options.num_levels - 1);
+        ASSERT_EQ(f.largest_seqno, expected_file_seqno);
+      }
+    }
+    ASSERT_TRUE(found);
+    ASSERT_EQ(Get("c"), "vc");
+
+    // The iterator predates the ingestion, so it sees b and d as one run of
+    // deleted keys.
+    attempted_insert_ranges_.clear();
+    if (Forward()) {
+      iter->SeekToFirst();
+      ASSERT_TRUE(iter->Valid());
+      ASSERT_EQ(iter->key(), "a");
+      iter->Next();
+      ASSERT_TRUE(iter->Valid());
+      ASSERT_EQ(iter->key(), "e");
+    } else {
+      iter->SeekToLast();
+      ASSERT_TRUE(iter->Valid());
+      ASSERT_EQ(iter->key(), "g");
+      iter->Prev();
+      ASSERT_TRUE(iter->Valid());
+      ASSERT_EQ(iter->key(), "e");
+      iter->Prev();
+      ASSERT_TRUE(iter->Valid());
+      ASSERT_EQ(iter->key(), "a");
+    }
+    ASSERT_OK(iter->status());
+    iter.reset();
+    ASSERT_EQ(attempted_insert_ranges_.size(), 1);
+    AssertRange(0, "b", "e");
+
+    EXPECT_EQ(Get("c"), "vc");
+    VerifyIteration({"a", "c", "e", "g"});
+    ASSERT_OK(Flush());
+    Reopen(options);
+    EXPECT_EQ(Get("c"), "vc");
+  }
+}
+
+// NewIterator() references the SuperVersion before it reads the sequence
+// number. An ingestion plus a write in between gives the iterator a sequence
+// number above the ingested files' while its SuperVersion lacks them. It must
+// not convert the deleted keys around those files into a range tombstone.
+TEST_P(ReadPathRangeTombstoneTest,
+       IngestedFileSurvivesIteratorWithOlderSuperVersion) {
+  for (bool new_seqno : {false, true}) {
+    SCOPED_TRACE(new_seqno ? "ingested with a new seqno"
+                           : "ingested with seqno zero");
+    Options options = CurrentOptions();
+    options.min_tombstones_for_range_conversion = 2;
+    options.disable_auto_compactions = true;
+    DestroyAndReopen(options);
+
+    for (const char* key : {"a", "b", "d", "e"}) {
+      ASSERT_OK(Put(key, std::string("v") + key));
+    }
+    ASSERT_OK(Flush());
+    ASSERT_OK(Delete("b"));
+    ASSERT_OK(Delete("d"));
+    // The memtable stays empty, so the ingestion does not switch it.
+    ASSERT_OK(Flush());
+
+    // "c" alone overlaps no key, so it gets seqno zero. Adding "b" overlaps
+    // the deleted "b" in L0, so the file gets a new seqno.
+    const std::string file = dbname_ + "/ingest.sst";
+    SstFileWriter writer(EnvOptions(), options);
+    ASSERT_OK(writer.Open(file));
+    if (new_seqno) {
+      ASSERT_OK(writer.Put("b", "vb2"));
+    }
+    ASSERT_OK(writer.Put("c", "vc"));
+    ASSERT_OK(writer.Finish());
+
+    SyncPoint::GetInstance()->LoadDependency(
+        {{"DBImpl::NewIterator:1",
+          "ReadPathRangeTombstoneTest::OlderSuperVersion:Ingest"},
+         {"ReadPathRangeTombstoneTest::OlderSuperVersion:Written",
+          "DBImpl::NewIterator:2"}});
+    std::unique_ptr<Iterator> iter;
+    port::Thread reader([&]() { iter.reset(db_->NewIterator(ReadOptions())); });
+    TEST_SYNC_POINT("ReadPathRangeTombstoneTest::OlderSuperVersion:Ingest");
+    ASSERT_OK(db_->IngestExternalFile({file}, IngestExternalFileOptions()));
+    ASSERT_OK(Put("z", "vz"));
+    TEST_SYNC_POINT("ReadPathRangeTombstoneTest::OlderSuperVersion:Written");
+    reader.join();
+    SyncPoint::GetInstance()->ClearTrace();
+    SyncPoint::GetInstance()->LoadDependency({});
+
+    std::vector<LiveFileMetaData> files;
+    db_->GetLiveFilesMetaData(&files);
+    bool found = false;
+    for (const auto& f : files) {
+      if (f.largestkey == "c") {
+        found = true;
+        ASSERT_EQ(f.largest_seqno > 0, new_seqno);
+      }
+    }
+    ASSERT_TRUE(found);
+
+    // The iterator reads past the write of "z" but not the ingested file.
+    attempted_insert_ranges_.clear();
+    std::vector<std::string> seen;
+    if (Forward()) {
+      for (iter->SeekToFirst(); iter->Valid(); iter->Next()) {
+        seen.push_back(iter->key().ToString());
+      }
+    } else {
+      for (iter->SeekToLast(); iter->Valid(); iter->Prev()) {
+        seen.insert(seen.begin(), iter->key().ToString());
+      }
+    }
+    ASSERT_OK(iter->status());
+    ASSERT_EQ(seen, std::vector<std::string>({"a", "e", "z"}));
+    iter.reset();
+    ASSERT_EQ(attempted_insert_ranges_.size(), 1);
+    AssertRange(0, "b", "e");
+
+    EXPECT_EQ(Get("c"), "vc");
+    EXPECT_EQ(Get("b"), new_seqno ? "vb2" : "NOT_FOUND");
+    ASSERT_OK(Flush());
+    Reopen(options);
+    EXPECT_EQ(Get("c"), "vc");
+  }
+}
+
+// One IngestExternalFiles() call installs files in two column families. Each
+// column family's older iterator must refuse its conversion.
+TEST_P(ReadPathRangeTombstoneTest, AtomicIngestionSurvivesOlderIterators) {
+  Options options = CurrentOptions();
+  options.min_tombstones_for_range_conversion = 2;
+  options.disable_auto_compactions = true;
+  DestroyAndReopen(options);
+  CreateAndReopenWithCF({"pikachu"}, options);
+  std::vector<std::string> files;
+  for (int cf = 0; cf < 2; cf++) {
+    files.push_back(dbname_ + "/ingest_c" + std::to_string(cf) + ".sst");
+    SstFileWriter writer(EnvOptions(), options);
+    ASSERT_OK(writer.Open(files.back()));
+    ASSERT_OK(writer.Put("c", "vc"));
+    ASSERT_OK(writer.Finish());
+    for (const char* key : {"a", "b", "d", "e"}) {
+      ASSERT_OK(Put(cf, key, std::string("v") + key));
+    }
+    ASSERT_OK(Flush(cf));
+    ASSERT_OK(Delete(cf, "b"));
+    ASSERT_OK(Delete(cf, "d"));
+  }
+  std::unique_ptr<Iterator> it0(db_->NewIterator(ReadOptions(), handles_[0]));
+  std::unique_ptr<Iterator> it1(db_->NewIterator(ReadOptions(), handles_[1]));
+  std::vector<IngestExternalFileArg> args(2);
+  for (int cf = 0; cf < 2; cf++) {
+    args[cf].column_family = handles_[cf];
+    args[cf].external_files = {files[cf]};
+  }
+  ASSERT_OK(db_->IngestExternalFiles(args));
+
+  attempted_insert_ranges_.clear();
+  for (auto* it : {it0.get(), it1.get()}) {
+    if (Forward()) {
+      for (it->SeekToFirst(); it->Valid(); it->Next()) {
+      }
+    } else {
+      for (it->SeekToLast(); it->Valid(); it->Prev()) {
+      }
+    }
+    ASSERT_OK(it->status());
+  }
+  it0.reset();
+  it1.reset();
+  EXPECT_EQ(attempted_insert_ranges_.size(), 2);
+  EXPECT_EQ(Get(0, "c"), "vc");
+  EXPECT_EQ(Get(1, "c"), "vc");
+}
+
+// An iterator created after an ingestion sees the ingested files, so the
+// barrier must still let it convert.
+TEST_P(ReadPathRangeTombstoneTest, IteratorCreatedAfterIngestionConverts) {
+  for (bool seqno_zero : {true, false}) {
+    SCOPED_TRACE(seqno_zero ? "ingested with seqno zero"
+                            : "ingested with a new seqno");
+    Options options = CurrentOptions();
+    options.min_tombstones_for_range_conversion = 2;
+    options.disable_auto_compactions = true;
+    options.statistics = CreateDBStatistics();
+    DestroyAndReopen(options);
+
+    // "x" alone overlaps no key, so it gets seqno zero. Adding "e" overlaps
+    // the flushed "e", so the file gets a new seqno.
+    const std::string file = dbname_ + "/ingest_x.sst";
+    SstFileWriter writer(EnvOptions(), options);
+    ASSERT_OK(writer.Open(file));
+    if (!seqno_zero) {
+      ASSERT_OK(writer.Put("e", "ve2"));
+    }
+    ASSERT_OK(writer.Put("x", "vx"));
+    ASSERT_OK(writer.Finish());
+    for (const char* key : {"a", "b", "c", "d", "e"}) {
+      ASSERT_OK(Put(key, std::string("v") + key));
+    }
+    ASSERT_OK(Flush());
+    ASSERT_OK(Delete("b"));
+    ASSERT_OK(Delete("c"));
+    ASSERT_OK(Delete("d"));
+    IngestExternalFileOptions ifo;
+    ifo.allow_blocking_flush = false;
+    ASSERT_OK(db_->IngestExternalFile({file}, ifo));
+
+    {
+      std::unique_ptr<Iterator> it(db_->NewIterator(ReadOptions()));
+      for (it->SeekToFirst(); it->Valid(); it->Next()) {
+      }
+      ASSERT_OK(it->status());
+    }
+    EXPECT_EQ(TestGetTickerCount(options, READ_PATH_RANGE_TOMBSTONES_INSERTED),
+              1);
+    VerifyIteration({"a", "e", "x"});
+    EXPECT_EQ(Get("e"), seqno_zero ? "ve" : "ve2");
+  }
+}
+
+// A conversion that runs before an ingestion leaves a range tombstone in the
+// memtable over the gap where the file lands. The ingestion must either fail
+// or make the file visible, whatever the ingestion options.
+TEST_P(ReadPathRangeTombstoneTest, IngestionAfterConversionIsNeverHidden) {
+  enum class IngestMode {
+    kNoGlobalSeqnoNoFlush,
+    kNoGlobalSeqno,
+    kGlobalSeqno,
+    kDbGenerated
+  };
+  for (IngestMode mode :
+       {IngestMode::kNoGlobalSeqnoNoFlush, IngestMode::kNoGlobalSeqno,
+        IngestMode::kGlobalSeqno, IngestMode::kDbGenerated}) {
+    SCOPED_TRACE("ingest mode " + std::to_string(static_cast<int>(mode)));
+    Options options = CurrentOptions();
+    options.min_tombstones_for_range_conversion = 2;
+    options.disable_auto_compactions = true;
+    options.statistics = CreateDBStatistics();
+    DestroyAndReopen(options);
+
+    std::string file = dbname_ + "/ingest_c.sst";
+    if (mode == IngestMode::kDbGenerated) {
+      const std::string src_path = dbname_ + "_src";
+      ASSERT_OK(DestroyDB(src_path, options));
+      std::unique_ptr<DB> src;
+      ASSERT_OK(DB::Open(options, src_path, &src));
+      ASSERT_OK(src->Put(WriteOptions(), "c", "vc"));
+      ASSERT_OK(src->Flush(FlushOptions()));
+      std::vector<LiveFileMetaData> src_files;
+      src->GetLiveFilesMetaData(&src_files);
+      ASSERT_EQ(src_files.size(), 1);
+      file = src_files[0].directory + "/" + src_files[0].relative_filename;
+      ASSERT_OK(src->Close());
+    } else {
+      SstFileWriter writer(EnvOptions(), options);
+      ASSERT_OK(writer.Open(file));
+      ASSERT_OK(writer.Put("c", "vc"));
+      ASSERT_OK(writer.Finish());
+    }
+
+    for (const char* key : {"a", "b", "d", "e"}) {
+      ASSERT_OK(Put(key, std::string("v") + key));
+    }
+    ASSERT_OK(Flush());
+    ASSERT_OK(Delete("b"));
+    ASSERT_OK(Delete("d"));
+    attempted_insert_ranges_.clear();
+    VerifyIteration({"a", "e"});
+    ASSERT_EQ(TestGetTickerCount(options, READ_PATH_RANGE_TOMBSTONES_INSERTED),
+              1);
+    AssertRange(0, "b", "e");
+
+    IngestExternalFileOptions ifo;
+    ifo.allow_global_seqno = mode == IngestMode::kGlobalSeqno;
+    ifo.allow_blocking_flush = mode != IngestMode::kNoGlobalSeqnoNoFlush;
+    ifo.allow_db_generated_files = mode == IngestMode::kDbGenerated;
+    Status s = db_->IngestExternalFile({file}, ifo);
+    if (mode == IngestMode::kGlobalSeqno) {
+      ASSERT_OK(s);
+    } else {
+      ASSERT_TRUE(s.IsInvalidArgument()) << s.ToString();
+    }
+
+    const std::string expected_c = s.ok() ? "vc" : "NOT_FOUND";
+    EXPECT_EQ(Get("c"), expected_c);
+    VerifyIteration(s.ok() ? std::vector<std::string>{"a", "c", "e"}
+                           : std::vector<std::string>{"a", "e"});
+    ASSERT_OK(Flush());
+    Reopen(options);
+    EXPECT_EQ(Get("c"), expected_c);
+  }
+}
+
 // SeekForPrev uses the reverse path: PrevInternal -> FindValueForCurrentKey.
 // When a key has more versions than max_sequential_skip_in_iterations,
 // FindValueForCurrentKey delegates to FindValueForCurrentKeyUsingSeek which

db/db_secondary_test.cc

@@ -933,6 +933,63 @@ TEST_F(DBSecondaryTest, OpenFilesAsync) {
   VerifySecondaryValue("key", "value3");
 }
 
+TEST_F(DBSecondaryTest, OpenFilesAsyncNonDefaultCfCatchUp) {
+  Options options = CurrentOptions();
+  options.disable_auto_compactions = true;
+  CreateAndReopenWithCF({"cf1"}, options);
+  // One open-time table in the non-default column family.
+  ASSERT_OK(Put(1, "key", "v1"));
+  ASSERT_OK(Flush(1));
+
+  Options secondary_options = options;
+  secondary_options.max_open_files = -1;
+  secondary_options.open_files_async = true;
+  secondary_options.skip_stats_update_on_db_open = true;
+
+  // Give the HIGH pool enough threads that holding the async opener back does
+  // not block the primary's own flushes.
+  const int original_high_threads =
+      env_->GetBackgroundThreads(Env::Priority::HIGH);
+  Defer restore_high_threads([&] {
+    env_->SetBackgroundThreads(original_high_threads, Env::Priority::HIGH);
+  });
+  env_->SetBackgroundThreads(4, Env::Priority::HIGH);
+
+  // Hold the background opener back so cf1's open-time table stays unpinned
+  // through catch-up.
+  SyncPoint::GetInstance()->LoadDependency(
+      {{"DBSecondaryTest::OpenFilesAsyncNonDefaultCfCatchUp:Release",
+        "DBImpl::BGWorkAsyncFileOpen::Start"},
+       {"DBImpl::BGWorkAsyncFileOpen:Done",
+        "DBSecondaryTest::OpenFilesAsyncNonDefaultCfCatchUp:Done"}});
+  SyncPoint::GetInstance()->EnableProcessing();
+  Defer cleanup_sync_point([] {
+    SyncPoint::GetInstance()->DisableProcessing();
+    SyncPoint::GetInstance()->ClearAllCallBacks();
+  });
+
+  OpenSecondaryWithColumnFamilies({"cf1"}, secondary_options);
+  ASSERT_EQ(2, handles_secondary_.size());
+
+  // The primary replaces cf1's open-time table and deletes it.
+  ASSERT_OK(Put(1, "key", "v2"));
+  ASSERT_OK(Flush(1));
+  CompactRangeOptions cro;
+  cro.bottommost_level_compaction = BottommostLevelCompaction::kForce;
+  ASSERT_OK(db_->CompactRange(cro, handles_[1], nullptr, nullptr));
+  ASSERT_OK(db_->FlushWAL(/*sync=*/true));
+
+  // Before the fix, catch-up re-opened cf1's compacted-away open-time table and
+  // failed with PathNotFound, latching the tailer.
+  ASSERT_OK(db_secondary_->TryCatchUpWithPrimary());
+  VerifySecondaryValue(handles_secondary_[1], "key", "v2");
+
+  // Let the held-back opener drain before teardown.
+  TEST_SYNC_POINT("DBSecondaryTest::OpenFilesAsyncNonDefaultCfCatchUp:Release");
+  TEST_SYNC_POINT("DBSecondaryTest::OpenFilesAsyncNonDefaultCfCatchUp:Done");
+  CloseSecondary();
+}
+
 TEST_F(DBSecondaryTest, OpenFilesAsyncRequiresSkipStatsUpdate) {
   Options options = CurrentOptions();
   options.open_files_async = true;

db/memtable.cc

@@ -972,7 +972,7 @@ void MemTable::ConstructFragmentedRangeTombstones() {
 
 bool MemTable::AddLogicallyRedundantRangeTombstone(
     SequenceNumber seq, const Slice& start_key, const Slice& end_key,
-    port::RWMutex& ingest_sst_lock) {
+    uint64_t version_number, port::RWMutex& ingest_sst_lock) {
   // Fast path: skip if already immutable.
   if (is_immutable_.LoadRelaxed()) {
     return false;
@@ -1016,6 +1016,11 @@ bool MemTable::AddLogicallyRedundantRangeTombstone(
   if (seq < ingest_seqno_barrier_.LoadRelaxed()) {
     return false;
   }
+  // The iterator's Version predates an ingestion, so the deleted run it saw
+  // can surround ingested keys that it cannot see.
+  if (version_number < ingest_version_barrier_.LoadRelaxed()) {
+    return false;
+  }
 
   MemTablePostProcessInfo post_process_info;
   Status s = Add(seq, kTypeRangeDeletion, start_key, end_key,
@@ -1034,6 +1039,12 @@ void MemTable::BumpIngestSeqnoBarrier(SequenceNumber y) {
   }
 }
 
+void MemTable::BumpIngestVersionBarrier(uint64_t version_number) {
+  if (ingest_version_barrier_.LoadRelaxed() < version_number) {
+    ingest_version_barrier_.StoreRelaxed(version_number);
+  }
+}
+
 port::RWMutex* MemTable::GetLock(const Slice& key) {
   return &locks_[GetSliceRangedNPHash(key, locks_.size())];
 }

db/memtable.h

@@ -337,10 +337,14 @@ class ReadOnlyMemTable {
   // - the per-memtable ingest seqno barrier already exceeds `seq` (an
   //   ingestion has committed an L0 file at a seq that this converted
   //   tombstone would shadow) or an ingestion is in progress.
+  // - `version_number`, the number of the Version the converting iterator
+  //   reads, is below the per-memtable ingest version barrier (the iterator
+  //   cannot see a file ingested while this memtable was active).
   // Returns true if the range tombstone was inserted, false if skipped.
   virtual bool AddLogicallyRedundantRangeTombstone(
       SequenceNumber /*seq*/, const Slice& /*start_key*/,
-      const Slice& /*end_key*/, port::RWMutex& /*ingest_sst_lock*/) {
+      const Slice& /*end_key*/, uint64_t /*version_number*/,
+      port::RWMutex& /*ingest_sst_lock*/) {
     return false;
   }
 
@@ -899,7 +903,7 @@ class MemTable final : public ReadOnlyMemTable {
 
   bool AddLogicallyRedundantRangeTombstone(
       SequenceNumber seq, const Slice& start_key, const Slice& end_key,
-      port::RWMutex& ingest_sst_lock) override;
+      uint64_t version_number, port::RWMutex& ingest_sst_lock) override;
 
   // Monotonically raises ingest_seqno_barrier_ to `y` (no-op if `y` is not
   // greater than the current value). The conversion's barrier check
@@ -911,6 +915,13 @@ class MemTable final : public ReadOnlyMemTable {
   // a CAS loop. Only IngestExternalFiles calls this.
   void BumpIngestSeqnoBarrier(SequenceNumber y);
 
+  // Monotonically raises ingest_version_barrier_ to `version_number`, the
+  // number of the Version that installed an ingestion.
+  //
+  // REQUIRES: DB mutex held by the caller. Only IngestExternalFiles calls
+  // this.
+  void BumpIngestVersionBarrier(uint64_t version_number);
+
   bool IsFragmentedRangeTombstonesConstructed() const override {
     return fragmented_range_tombstone_list_.get() != nullptr ||
            is_range_del_table_empty_.LoadRelaxed();
@@ -975,6 +986,10 @@ class MemTable final : public ReadOnlyMemTable {
   // ColumnFamilyData::ingest_sst_lock_.
   RelaxedAtomic<SequenceNumber> ingest_seqno_barrier_{0};
 
+  // Number of the latest Version that installed an ingestion. See also
+  // ColumnFamilyData::ingest_sst_lock_.
+  RelaxedAtomic<uint64_t> ingest_version_barrier_{0};
+
   SequenceNumber creation_seq_;
 
   // the earliest log containing a prepared section

db/version_edit_handler.cc

@@ -934,8 +934,14 @@ Status VersionEditHandlerPointInTime::MaybeCreateVersionBeforeApplyEdit(
         if (negative_edge) {
           builder->RedoLastApply();
         }
-        if (s.IsCorruption()) {
-          // This point in time cannot be recovered; skip it and continue.
+        if (s.IsCorruption() ||
+            (s.IsPathNotFound() && no_error_if_files_missing_)) {
+          // This point in time cannot be recovered; skip it and continue. A
+          // PathNotFound is a file the primary compacted away before this
+          // handler opened it, which open_files_async makes reachable during a
+          // secondary's catch-up by leaving recovery-time files unpinned until
+          // the background opener reaches them. Treat it as missing rather than
+          // latching the error, the same tolerance LoadTables() applies.
           s = Status::OK();
         }
         builder->CommitLastApply();

db/wide/db_wide_blob_direct_write_test.cc

@@ -750,8 +750,9 @@ TEST_F(DBWideBlobDirectWriteTest, DirectWriteWideEntityBeforeAndAfterFlush) {
 
   std::vector<BlobFileAddition> additions;
   std::vector<BlobFileGarbage> garbages;
-  ASSERT_OK(mgr->PrepareFlushAdditions(WriteOptions(), /*num_generations=*/1,
-                                       &additions, &garbages));
+  ASSERT_OK(mgr->PrepareFlushAdditions(
+      WriteOptions(), {cfh->cfd()->imm()->GetEarliestMemTableID()}, &additions,
+      &garbages));
   ASSERT_FALSE(additions.empty());
   ASSERT_TRUE(garbages.empty());
 
@@ -1033,8 +1034,9 @@ TEST_F(DBWideBlobDirectWriteTest,
 
   std::vector<BlobFileAddition> additions;
   std::vector<BlobFileGarbage> garbages;
-  ASSERT_OK(mgr->PrepareFlushAdditions(WriteOptions(), /*num_generations=*/1,
-                                       &additions, &garbages));
+  ASSERT_OK(mgr->PrepareFlushAdditions(
+      WriteOptions(), {cfh->cfd()->imm()->GetEarliestMemTableID()}, &additions,
+      &garbages));
   ASSERT_EQ(additions.size(), 1U);
   ASSERT_TRUE(garbages.empty());
 

include/rocksdb/advanced_options.h

@@ -1427,6 +1427,7 @@ struct AdvancedColumnFamilyOptions {
   // * prefix extractor reads that are neither total-order
   //   (ReadOptions::total_order_seek / ReadOptions::auto_prefix_mode) nor
   //   bounded by ReadOptions::prefix_same_as_start
+  // * reads with ReadOptions::read_tier other than kReadAllTier
   //
   // Even if the above restrictions are met, there are still scenarios where a
   // converted range tombstone may be discarded:
@@ -1439,7 +1440,8 @@ struct AdvancedColumnFamilyOptions {
   //     and the snapshot seq is at or above its min uncommitted seq.
   //   * An IngestExternalFile call is currently in flight on this column
   //     family OR the inserted range tombstone seqno would be lower than the
-  //     ingested file seqno.
+  //     ingested file seqno OR the iterator's Version predates a file
+  //     ingested while the active memtable was current.
   //
   // Read-write iterators using ReadOptions::table_filter are rejected while
   // this option is enabled, see more details in ReadOptions::table_filter

include/rocksdb/version.h

@@ -13,7 +13,7 @@
 // minor or major version number planned for release.
 #define ROCKSDB_MAJOR 11
 #define ROCKSDB_MINOR 13
-#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)

unreleased_history/behavior_changes/no_auto_resume_stops_writes.md

@@ -1 +0,0 @@
-With `max_bgerror_resume_count <= 0` (auto resume disabled), a retryable or file-scoped IO error during compaction or flush is now a hard background error that stops writes until `DB::Resume()` is called. Previously, compaction errors were silently retried, and flush errors with a healthy WAL were soft errors that let writes continue while flushes were stopped.

unreleased_history/bug_fixes/delete_scheduler_unlinked_trash_accounting.md

@@ -1 +0,0 @@
-Fixed two more `SstFileManager` trash accounting leaks: retrying a failed trash deletion immediately (for example with rate limiting disabled) after the file had disappeared, and a trash file that was deleted but whose directory sync then failed. In both cases the file's size stayed in the total trash size and tracked size until restart.

unreleased_history/bug_fixes/delete_scheduler_vanished_trash_accounting.md

@@ -1 +0,0 @@
-Fixed `SstFileManager` trash accounting leaking the size of a queued trash file that disappeared before deletion (for example because its directory was renamed). The leaked bytes stayed in the total trash size until restart, which could push every later deletion past `max_trash_db_ratio` and make it immediate instead of rate limited.

unreleased_history/bug_fixes/multiget_async_io_fallback.md

@@ -1 +0,0 @@
-Fixed `MultiGet` with `ReadOptions::async_io` returning `NotSupported` when io_uring initialization or an asynchronous read submission fails. It now checks async I/O support on the reading thread and falls back to synchronous reads when needed.

unreleased_history/bug_fixes/subset_checkpoint_options_cf_membership.md

@@ -1,5 +0,0 @@
-Fixed subset checkpoints so their newest OPTIONS file lists only the column
-families retained by the checkpoint. This applies with both legacy
-highest-numbered OPTIONS selection and MANIFEST-tracked OPTIONS selection, and
-prevents `LoadLatestOptions` from recreating column families that the subset
-checkpoint intentionally dropped from its copied MANIFEST.

unreleased_history/new_features/fs_is_preallocation_supported.md

@@ -1 +0,0 @@
-Added `FileSystem::IsPreallocationSupported()`. A `FileSystem` that never preallocates file space can return false so that DB open skips reopening and truncating the last WAL file.

unreleased_history/performance_improvements/perf_context_tls_init.md

@@ -1 +0,0 @@
-Perf context counters and timers no longer call the thread-local init function of `perf_context` on every update. With `kEnableCount`, a counter update costs about 60% less.

unreleased_history/public_api_changes/committed_options_manifest_number.md

@@ -1,11 +0,0 @@
-Added the experimental `DBOptions::track_options_file_number_in_manifest`
-rollout option (with C and Java bindings). The option is persisted in each
-OPTIONS file. When enabled, RocksDB durably creates an OPTIONS file and then
-commits its file number in MANIFEST. Column-family create/drop commits the
-configuration snapshot and membership in one MANIFEST record. A newer tracked
-OPTIONS file without a commit is ignored after a crash, while a newer OPTIONS
-file written with tracking disabled retains the legacy highest-numbered-file
-behavior. Batched column-family changes publish one snapshot; after partial
-success it can be a safe superset that persisted-options APIs filter against
-live MANIFEST membership. Ordinary `DB::Open` continues to use caller-supplied
-options and does not depend on the operational snapshot.

unreleased_history/public_api_changes/encode_entry.md

@@ -1 +0,0 @@
-Added `EncodeEntry()`, `PackSequenceAndType()`, and `UnPackSequenceAndType()` in `rocksdb/utilities/types_util.h` to construct internal keys and pack or unpack their sequence numbers and entry types. `kInternalKeyFooterSize` exposes the footer size in bytes.

unreleased_history/public_api_changes/ldb_rebind_options_manifest.md

@@ -1 +0,0 @@
-Added `ldb update_manifest --rebind_options_file --options_file=OPTIONS-<number>` for offline repair of a committed MANIFEST OPTIONS file number. Operators must select a trusted snapshot that has MANIFEST tracking enabled and provide any external DB/CF allocation roots that are not serialized in OPTIONS.