Don't let error recovery wait forever behind a writer stalled on the failed flush (#15605)

Summary:
Pull Request resolved: https://github.com/meta/rocksdb/pull/15605

**Problem:** automatic error recovery can hang forever, with all writes stopped, when a writer is waiting on a write stop (for example the `max_write_buffer_number` memtable limit) as the flush that would end it fails. Only a restart clears it.

**Why:** the stalled writer leads the write queue. It waits in `DelayWrite()` for the stop to end, and keeps waiting while recovery is in progress, so any recovery step that enters the write queue waits behind it. Two such steps exist:
1. A retryable SST write error is a soft error, and recovery retries the flush of the immutable memtables, which does not need the write queue. But with `two_write_queues = true`, `ResumeImpl()` first enters both write queues to sync `LastSequence()` with the allocated sequence numbers.
2. A hard error with auto recovery, for example a retryable error syncing the closed WALs during the flush of a DB with two column families, is recovered by an atomic flush of all column families. Its memtable switch enters the write queue in every configuration. New writes already fail with the background error at this point (`PreprocessWrite()` checks `IsDBStopped()`); only a writer that was already waiting keeps waiting.

**Fix:**
- The sequence sync at recovery start no longer waits for a write queue whose leader is in a write stall. That leader stalls before it allocates sequence numbers, and it cannot leave the stall while recovery holds the DB mutex, so the sync stays exact. If the stall ends while recovery waits for the other queue, the sync retries. Queues without a stall are still drained first, and the recovery flushes that switch memtables keep their own full fence.
- `DelayWrite()` stops waiting on a write stop once the DB is stopped, and returns the background error, like a new write does.

**Tests:**
- In `error_handler_fs_test`:
  - `RetryableFlushErrorWithWriterStalledOnFlush` (with and without `two_write_queues`): the first flush fails with a retryable SST write error after the writer stops on its memtables; the writer finishes after recovery.
  - `FlushWALSyncErrorWithWriterStalledOnFlush`: with two column families, the closed WAL sync fails after the writer stops; the writer gets the hard error, recovery completes, and the next write succeeds.
- In `write_prepared_transaction_seqno_test`:
  - `RecoverySeqnoSyncRetriesWhenStallEndsDuringNonmemJoin`: the main queue stall ends while recovery joins the nonmem queue, and the writer allocates before recovery syncs; recovery retries and syncs only after the writer publishes.
  - `RecoverySeqnoSyncSkipsStalledWalOnlyCommit`: a WAL-only commit is stalled on the nonmem queue; recovery syncs and finishes while the commit still waits.

Reviewed By: archang19

Differential Revision: D127129804

fbshipit-source-id: 88ed4938be39d7222ceb7c28437187989885b5d0

HISTORY.md

@@ -1,6 +1,81 @@
 # Rocksdb Change Log
 > NOTE: Entries for next release do not go here. Follow instructions in `unreleased_history/README.txt`
 
+## 11.14.0 (11/05/2026)
+### New Features
+* Added experimental `SetAsyncReadCPUTimeEnabled()`: the next async or coroutine read started on the thread records its CPU time in `PerfContext::get_cpu_nanos` at any perf level except `kDisable`, without turning on the other timers of `kEnableTimeAndCPUTimeExceptForMutex`.
+* Added `SplitDB()`, `DeleteSplitRanges()`, and `MergeDB()` in `rocksdb/utilities/db_split_merge.h`, plus a `db_split_merge` command-line tool, for moving selected column-family key ranges into a new DB and copying ranges into existing column families.
+* Added opt-in sampled logging of long-held DB-wide mutexes under `StatsLevel::kAll`, with a runtime-changeable `mutex_hold_log_threshold_us` DB option defaulting to 0 (disabled).
+* Add experimental live tuning of a DB's options (`StartLiveTuneCycle()`, `GetLiveTuneStatus()`, `StopLiveTuneCycle()` in `include/rocksdb/live_tune.h`): one call runs a cycle that measures the DB, asks an application-provided `LiveTuneAdvisor` for option changes, applies them, and reverts them when they break the application's constraints or do not prove better.
+* Added `CompactionFilter::GetMergeOrder()` and `CompactionFilterFactory::GetMergeOrder()` to optionally run the merge operator before the compaction filter during compaction, so that the filter sees merge results. The default behavior is unchanged.
+* Parallel compression (`CompressionOptions::parallel_threads > 1`) now works with a `user_defined_index_factory` whose builders support it, including the built-in trie index. Previously a custom index always forced single-threaded compression.
+* Added option `reject_too_old_file` to `IngestExternalFileOptions` and `ImportColumnFamilyOptions` to reject files whose data is already older than the column family's `periodic_compaction_seconds` (or `ttl` for FIFO compaction).
+* Added BlobDB support for remote compaction. Blob garbage collection uses the
+cutoff computed by the primary; workers receiving legacy input without a cutoff
+safely disable blob garbage collection for that job. When the primary rejects a
+remote blob result because its metadata, files, or capabilities fail validation,
+it reruns the compaction locally, so `CompactRange` and `CompactFiles` succeed.
+`CompactionService` `kFailure` semantics are unchanged; ZippyDB production also
+maps service failures to local fallback through `HandleFailure(FALLBACK_LOCAL)`.
+* `CompactRangeOptions::skip_missing_blob_files` enables offline repair that deletes values referencing integrated external blob files absent from Version metadata, preserves available wide entity columns, and rewrites selected SSTs locally, bypassing `DBOptions::compaction_service`. It defaults to false; FIFO compaction and legacy BlobDB are unsupported.
+* Added `db_ownership_start_time` and `data_file_provenance` (decoded by `DataFileProvenance`) to `SstFileMetaData`, recording when the DB began holding each SST file and how it got the file (e.g. flush, compaction, ingestion, or import).
+* Added `TableFactory::IsTableBuilderSupported()` and `ExternalTableFactory::IsTableBuilderSupported()` for table factories that only read ingested or imported files. A column family whose factory returns false must set `disallow_memtable_writes`, is never scheduled for automatic compaction, does not stall writes for L0 files or pending compaction bytes, and returns `NotSupported` from `CompactRange()` and `CompactFiles()` instead of failing a background compaction.
+* trace_analyzer: new `--analyze_put_entity` flag analyzes PutEntity queries (previously their statistics were neither enabled nor disabled explicitly).
+* Added `TraceOptions::include_read_write_options`: query traces can record the main `ReadOptions` and `WriteOptions` of each traced call (e.g. `fill_cache`, `disableWAL`), which replay then uses instead of default options.
+
+### Public API Changes
+* Added `GetEncodedEntrySize()` and an `EncodeEntry()` overload that writes internal keys into caller-owned buffers without allocating on success.
+* Added `Env::Priority::READ` and `ThreadStatus::READ_PRIORITY` for a dedicated Env-managed read-work thread pool. `Env::Priority::USER` remains reserved for internal statistics and is not backed by a thread pool.
+* Added experimental `MultiScanArgs::use_parallel_async_io_block_decompression` for materializing asynchronously prefetched data blocks, including decompression, in parallel on the `Env::Priority::READ` thread pool. A zero-sized pool preserves inline materialization.
+* Added the experimental `IndexFactory` interface (`include/rocksdb/index_factory.h`) for pluggable SST indexes; the `UserDefinedIndex*` names remain as aliases. `BlockBasedTableOptions::index_mode` chooses which indexes an SST is built with and which one reads use by default, superseding the now-deprecated `use_udi_as_primary_index` and `fail_if_no_udi_on_open`. `ReadOptions::read_index` overrides that choice per read, and the `standard_index_is_stub` table property marks SSTs built with `kCustomOnly`. Both options are also available in the C API.
+* Added experimental `DB::GetEntityWithMetadata()` and `DB::MultiGetEntityWithMetadata()`, the wide-column entity counterparts of `DB::GetWithMetadata()` and `DB::MultiGetWithMetadata()` with opt-in explicit-snapshot newer-version detection. Requesting key timestamps returns `NotSupported`.
+* Removed support for building on Solaris, which had not compiled since 5.18. Removed RocksJava `Environment.isSolaris()`.
+
+### Behavior Changes
+* With `ReadOptions::async_io`, `MultiGet` now reads data blocks of direct-IO files with asynchronous reads (`FSRandomAccessFile::ReadAsync`) instead of a synchronous `MultiRead`. Each block read is aligned separately rather than merged with adjacent reads. The bytes read are still charged to `DBOptions::rate_limiter` at `ReadOptions::rate_limiter_priority`.
+* Asynchronous file reads (`FSRandomAccessFile::ReadAsync`, used with `ReadOptions::async_io` by `MultiGet` and iterator prefetching) are now charged to `DBOptions::rate_limiter` at `ReadOptions::rate_limiter_priority`, like synchronous reads. A read is charged once the file system accepts it.
+* `CreateColumnFamily()` and `CreateColumnFamilies()` now run the table factory option validation that `DB::Open()` runs, so they return `InvalidArgument` instead of creating a column family that the next `DB::Open()` would reject. External table factories wrapped by `NewExternalTableFactory()` now receive `ValidateOptions()` calls.
+* With `CompactionOptionsFIFO::allow_trivial_copy_when_change_temperature`, each file copied for a temperature change is now reported to `EventListener::OnTableFileCreationStarted()` and `OnTableFileCreated()` (with the original file's table properties) and included in `CompactionJobInfo::table_properties`.
+* `DB::GetMergeOperands()` now resets only the `merge_operands` entries it fills, instead of all `expected_max_number_of_operands` entries. Entries past `*number_of_operands`, and the whole array on a non-OK result, are left as the caller passed them.
+* The JSON written to the info LOG by `EventLogger` (`EVENT_LOG_v1` lines) and by `VersionEdit::DebugJSON()` now escapes special characters in keys and string values, and writes non-finite floating-point values as `null`, so the output is always valid JSON.
+* Improved how `periodic_compaction_phase_recovery_percent` spreads catch-up periodic compactions after it is enabled or changed (or the DB is reopened). This reduces a transient burst in periodic compactions when many DBs make such a change at once.
+* The new DB option `use_separate_remote_compaction_pool` defaults to false and can be changed through `SetDBOptions()` for newly scheduled jobs. When enabled with CompactionService, remote jobs use a dedicated pool and each local subcompaction uses LOW/BOTTOM and counts against the DB compaction limit. Existing jobs finish using their original scheduling model.
+* `DB::SetOptions()` now returns OK without writing a new OPTIONS file or installing a new SuperVersion when the call changes no option value, as `DB::SetDBOptions()` already does. Calls that set an object option, such as `table_factory` or one of its fields, are still applied in full. After a failed OPTIONS file write, the next `SetOptions()` or `SetDBOptions()` call writes the file even if no option value changes.
+* Custom index factories are rejected with user-defined timestamps in every index mode except `kStandardOnly`, which ignores the factory. Custom-index block-cache hits, misses, and adds now contribute to `BLOCK_CACHE_INDEX_*` instead of `BLOCK_CACHE_DATA_*`.
+Trie index construction returns `NotSupported` when a data-block offset or size exceeds `UINT32_MAX`, including when the trie is optional in `kStandardDefault`. Flush, compaction, and `SstFileWriter::Finish()` can therefore fail for such output; an SST larger than 4 GiB is not itself the rejection condition. File-size targets do not split L0 outputs or `SstFileWriter` outputs. Bound those outputs or disable trie construction with `kStandardOnly`.
+
+### Bug Fixes
+* Fixed a race where `IngestExternalFiles()` with `atomic_replace_range` could delete an existing SST file that a concurrently picked compaction (e.g. a universal compaction outputting to L0) used as input, causing the compaction to fail with "Corruption: VersionBuilder: Cannot delete table file ... since it is not in the LSM tree" and putting the DB into read-only mode. Atomic range replacement now returns `NotSupported` with FIFO compaction.
+* Fixed blob direct write when sealing a blob file fails during a flush or a write: the flush left its memtables marked as being flushed, so no later flush picked them up and error recovery waited for them forever, and the blob file was dropped from tracking, so a retried flush could install table files whose blob file was never added to the MANIFEST.
+* 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.
+* Fixed blob direct write when the `SstFileManager` space limit is reached while a blob file is sealed: the sealed blob file was deleted while memtables still referenced it, so reads of its values failed and the values were lost, and reporting the error could deadlock with a concurrent memtable switch. The seal and the write or flush now succeed and keep the file, and writes then stop with a `SpaceLimit` background error, so for blob direct write the limit is soft: usage can exceed it by up to about `blob_file_size` × `blob_direct_write_partitions` for each memtable not flushed yet. Also fixed a write batch that failed after writing some of its values to blob files: those blobs are now counted as garbage, so their blob file can still be deleted.
+* Values that retain external blob references now keep their original timestamps during history compaction, preventing blob key verification failures.
+* Fixed `DB::ClipColumnFamily()` leaving an empty-string key outside the kept range when it is the only key in an L0 SST file.
+* Fixed `format_version=8` with `ReverseBytewiseComparator`: a block's stored common key prefix could extend past a user key (one following longer user keys that extend it), so reads and iteration could miss that key (or its deletion, exposing an older value), and compactions could fail with `Corruption`. Files written by affected versions (11.10 through 11.13) are now read correctly, and new files no longer have this problem.
+* Fixed SST file times that could defer periodic compaction and TTL. Files written by `SstFileWriter`, flushed during WAL recovery, or recovered by `RepairDB` now record a file creation time. Files added by `IngestExternalFile()` or `CreateColumnFamilyWithImport()` now keep the creation and oldest ancestor times from the file or export metadata (capped at the current time) instead of taking the time they were added, so ingesting or importing old data can make it due for periodic compaction or TTL soon afterward. FIFO TTL compaction now also deletes ingested `SstFileWriter` files, which previously blocked TTL deletion of themselves and all newer files. Files copied by FIFO compaction for a temperature change now keep the original file's creation time.
+* Fixed a debug-build assertion failure and incorrect cache charge and block-cache byte statistics in release builds when a full, strict-capacity LRU cache with metadata charging creates an uncharged standalone handle, including during secondary-cache promotion.
+* Fixed atomic flushes of all column families when another column family is dropped while they wait: `CompactRange()` no longer fails with a "Column family dropped" error, and `WaitForCompact()` and DB close no longer return OK without flushing the remaining column families.
+* Fixed a bug where MemPurge returned with the DB mutex unlocked when the compaction filter created for flush did not ignore snapshots.
+* Fixed a crash in MultiScan `Iterator::Prepare()` with a single scan range without a limit when the DB has L0 files, and an assertion failure in debug builds when `Prepare()` is called on an already-positioned iterator. `Prepare()` now leaves the iterator invalid until it is positioned again (e.g. by `Seek()`).
+* Fixed `MultiGet` returning NotFound for existing keys that are outside the prefix extractor's domain, with partitioned filters, `whole_key_filtering=false`, and filter partitions cut together with index partitions (`decouple_partitioned_filters=false`, also the layout of files written by older versions). A filter partition holding only such keys is empty, and `MultiGet` treated every key mapped to it as absent. `Get` was not affected.
+* Fixed prepared external file ingestions retaining unrelated obsolete SSTs until commit or abort. Prepared ingestions now protect only their staged files from deletion.
+* 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`.
+* Fixed a race with `recycle_log_file_num > 0` where an obsolete WAL could be reused for a new WAL while its old writer was still open. Closing the old writer then truncated the reused file to the old WAL's size, so writes in the new WAL past that point, including acknowledged `sync` writes, were lost on the next reopen.
+* Fixed remote compaction ignoring `preclude_last_level_data_seconds` and `preserve_internal_time_seconds`, which caused recent data to be written to the last level and recent sequence numbers to be zeroed. Takes effect once both the primary and the compaction worker are upgraded.
+* Fixed `DB::OpenAndCompact()` with `OpenAndCompactOptions::allow_resumption` not calling `EventListener::OnSubcompactionCompleted()` when the compaction fails to resume from its saved progress.
+* Fixed MANIFEST catch-up on secondary instances (`TryCatchUpWithPrimary()`) and follower instances installing a column family with no files and still returning OK. This happened when catch-up switched to a new MANIFEST and the primary deleted a file that MANIFEST lists before catch-up checked it, recording the deletion in a newer MANIFEST. Catch-up now keeps the column family's previous view until a complete state from the new MANIFEST is available.
+* Custom index blocks now reuse cached contents on SST reopen, and deprecated UDI boolean options can be cleared without overriding an explicitly configured index mode.
+Legacy UDI options copied from a factory can now be cleared before constructing a new factory without retaining the derived mode. The C API preserves both that rollback behavior and explicitly selected modes. OPTIONS files serialize equivalent legacy booleans for `kStandardDefault`, `kStandardRequired`, and `kCustomDefault` so older readers retain routing.
+Empty custom-index output retains the standard-index fallback in modes that write both indexes; custom-only output must remain nonempty. Default primary reads retain the table factory's precedence over a legacy per-read factory pointer, while explicit custom selectors still check its name.
+* Preserve legacy UDI routing when loading old OPTIONS files, and reject invalid index modes in standalone SST readers and writers. Table readers now charge the custom index blocks they own; a custom reader whose `ApproximateMemoryUsage()` already includes that block should override `IndexFactoryReader::MemoryUsageIncludesIndexBlock()` to return true so the block is not charged twice.
+Keep cached custom-index Slice views intact when a reader consumes its input. Register custom-index cache helpers without requiring a later change. Trie index size estimates now account for shared prefixes, preventing premature compaction file cuts, and reject data-block offsets or sizes exceeding the 32-bit on-disk limit instead of truncating them.
+
+### Performance Improvements
+* With `ReadOptions::pin_data`, iterators over read-only DBs opened with `max_open_files = -1` and `allow_mmap_reads = true` no longer accumulate memory as they move across SST files.
+* On a read-write primary, `DB::GetSnapshot()` and `DB::ReleaseSnapshot()` no longer acquire the DB mutex in the common case. Secondary and follower snapshot creation retains the DB mutex needed to synchronize with view installation, while release remains sharded. The snapshot list is now split into 16 per-CPU shards, each with its own lock, so primary threads creating and releasing snapshots no longer serialize on one lock or wait behind flush, compaction, or other work holding the DB mutex. Snapshot-release maintenance is deferred to existing flush, compaction, and periodic maintenance paths. Primary snapshot creation serializes with the DB-mutex-held phases of external file ingestion but can proceed while the MANIFEST write temporarily releases that mutex.
+
+
 ## 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.

db/compaction/compaction_outputs.cc

@@ -235,6 +235,32 @@ uint64_t CompactionOutputs::GetCurrentKeyGrandparentOverlappedBytes(
   return overlapped_bytes;
 }
 
+bool CompactionOutputs::PartitionerRequiresCut(
+    const CompactionIterator& c_iter) const {
+  assert(HasBuilder());
+  if (!partitioner_) {
+    return false;
+  }
+  if (partitioner_prefix_valid_) {
+    // Fast path: no per-key copy or virtual call. A partition is requested at
+    // the first key that no longer shares the established prefix. An empty
+    // prefix means no further partitions (every key "starts with" it).
+    return UNLIKELY(!c_iter.user_key().starts_with(partitioner_prefix_));
+  }
+  // Slow path: per-key callback (and copy elsewhere)
+  return UNLIKELY(partitioner_->ShouldPartition(PartitionerRequest(
+                      last_key_for_partitioner_, c_iter.user_key(),
+                      current_output_file_size_)) == kRequired);
+}
+
+bool CompactionOutputs::LargestUserKeyEquals(const Slice& user_key) {
+  const InternalKey& largest = current_output().meta.largest;
+  return largest.size() > 0 &&
+         compaction_->column_family_data()
+             ->user_comparator()
+             ->EqualWithoutTimestamp(largest.user_key(), user_key);
+}
+
 bool CompactionOutputs::ShouldStopBefore(const CompactionIterator& c_iter) {
   assert(c_iter.Valid());
   const Slice& internal_key = c_iter.key();
@@ -270,22 +296,8 @@ bool CompactionOutputs::ShouldStopBefore(const CompactionIterator& c_iter) {
     return true;
   }
 
-  if (partitioner_) {
-    if (partitioner_prefix_valid_) {
-      // Fast path: no per-key copy or virtual call. A partition is requested at
-      // the first key that no longer shares the established prefix. An empty
-      // prefix means no further partitions (every key "starts with" it).
-      if (UNLIKELY(!c_iter.user_key().starts_with(partitioner_prefix_))) {
-        return true;
-      }
-    } else {
-      // Slow path: per-key callback (and copy elsewhere)
-      if (UNLIKELY(partitioner_->ShouldPartition(PartitionerRequest(
-                       last_key_for_partitioner_, c_iter.user_key(),
-                       current_output_file_size_)) == kRequired)) {
-        return true;
-      }
-    }
+  if (PartitionerRequiresCut(c_iter)) {
+    return true;
   }
 
   // files output to Level 0 won't be split
@@ -400,13 +412,24 @@ Status CompactionOutputs::AddToOutput(
     const ParsedInternalKey& prev_iter_output_internal_key) {
   Status s;
   bool is_range_del = c_iter.IsDeleteRangeSentinelKey();
+  bool cut_before;
   if (is_range_del && compaction_->bottommost_level()) {
-    // We don't consider range tombstone for bottommost level since:
-    // 1. there is no grandparent and hence no overlap to consider
-    // 2. range tombstone may be dropped at bottommost level.
-    return s;
+    // At the bottommost level, a range tombstone is not by itself a reason to
+    // cut: there are no grandparents to limit overlap with, and the tombstone
+    // might be dropped. However, a boundary required by the SstPartitioner
+    // must still be honored; otherwise AddRangeDels() would extend the
+    // current file across it. No boundary can fall inside one user key, and
+    // a point key there may have its seqno zeroed, so that it sorts after a
+    // tombstone starting at the same user key.
+    if (!HasBuilder() || LargestUserKeyEquals(c_iter.user_key()) ||
+        !PartitionerRequiresCut(c_iter)) {
+      return s;
+    }
+    cut_before = true;
+  } else {
+    cut_before = ShouldStopBefore(c_iter) && HasBuilder();
   }
-  if (ShouldStopBefore(c_iter) && HasBuilder()) {
+  if (cut_before) {
     const Slice next_table_min_key = c_iter.key();
     s = close_file_func(c_iter.InputStatus(), prev_iter_output_internal_key,
                         next_table_min_key, &c_iter,

db/compaction/compaction_outputs.h

@@ -245,6 +245,15 @@ class CompactionOutputs {
   // before processing the current key in compaction iterator.
   bool ShouldStopBefore(const CompactionIterator& c_iter);
 
+  // Returns true iff the SST partitioner requires a new output file to start
+  // at the current key. Only meaningful while an output file is open: the
+  // partitioner state describes the current file.
+  bool PartitionerRequiresCut(const CompactionIterator& c_iter) const;
+
+  // Returns true iff the current output file has a largest key and its user
+  // key equals `user_key`, ignoring timestamps.
+  bool LargestUserKeyEquals(const Slice& user_key);
+
   // Re-establishes per-file state that depends on the first key accepted by a
   // freshly opened builder. This keeps caller-requested file cuts equivalent to
   // RocksDB's built-in file-boundary decisions for partition tracking.

db/compaction/tiered_compaction_test.cc

@@ -843,19 +843,20 @@ TEST_F(TieredCompactionTest, LevelOutofBoundaryRangeDelete) {
   ASSERT_OK(Flush());
 
   ASSERT_OK(db_->CompactRange(cro, nullptr, nullptr));
-  ASSERT_EQ("0,2,10",
-            FilesPerLevel());  // one file is at the proximal level
-                               // which only contains a range delete
+  ASSERT_EQ("0,3,10",
+            FilesPerLevel());  // the partitioner cuts before and after the
+                               // range delete, so it has a file of its own
+                               // at the proximal level
   std::vector<LiveFileMetaData> live_file_meta;
   db_->GetLiveFilesMetaData(&live_file_meta);
   bool found_sst_with_del = false;
   uint64_t sst_with_del_num = 0;
   for (const auto& meta : live_file_meta) {
     if (meta.num_deletions > 0) {
-      // found SST with del, which has 2 entries, one for data one for range del
+      // found SST with del, which only contains the range del
       ASSERT_EQ(meta.level,
                 kNumLevels - 2);  // output to proximal level
-      ASSERT_EQ(meta.num_entries, 2);
+      ASSERT_EQ(meta.num_entries, 1);
       ASSERT_EQ(meta.num_deletions, 1);
       found_sst_with_del = true;
       sst_with_del_num = meta.file_number;
@@ -974,7 +975,9 @@ TEST_F(TieredCompactionTest, UniversalRangeDelete) {
   ASSERT_OK(Flush());
 
   ASSERT_OK(db_->CompactRange(cro, nullptr, nullptr));
-  ASSERT_EQ("0,0,0,0,0,2,8", FilesPerLevel());
+  // the partitioner cuts before and after the range del, so it has a file of
+  // its own at the proximal level
+  ASSERT_EQ("0,0,0,0,0,3,8", FilesPerLevel());
   // find the SST with range del
   std::vector<LiveFileMetaData> live_file_meta;
   db_->GetLiveFilesMetaData(&live_file_meta);
@@ -982,10 +985,10 @@ TEST_F(TieredCompactionTest, UniversalRangeDelete) {
   uint64_t sst_with_del_num = 0;
   for (const auto& meta : live_file_meta) {
     if (meta.num_deletions > 0) {
-      // found SST with del, which has 2 entries, one for data one for range del
+      // found SST with del, which only contains the range del
       ASSERT_EQ(meta.level,
                 kNumLevels - 2);  // output_to_proximal_level level
-      ASSERT_EQ(meta.num_entries, 2);
+      ASSERT_EQ(meta.num_entries, 1);
       ASSERT_EQ(meta.num_deletions, 1);
       found_sst_with_del = true;
       sst_with_del_num = meta.file_number;

db/db_impl/db_impl.cc

@@ -354,6 +354,47 @@ Status DBImpl::Resume() {
   return s;
 }
 
+void DBImpl::SyncLastSequenceWithAllocatedAtRecoveryStart() {
+  mutex_.AssertHeld();
+  assert(two_write_queues_);
+  for (;;) {
+    WriteThread::Writer w;
+    const bool entered = write_thread_.EnterUnbatchedNonBlocking(&w, &mutex_);
+    const uint64_t stall =
+        entered ? 0 : write_thread_.GetBegunCountOfOutstandingStall();
+    WriteThread::Writer nonmem_w;
+    const bool nonmem_entered =
+        nonmem_write_thread_.EnterUnbatchedNonBlocking(&nonmem_w, &mutex_);
+    const uint64_t nonmem_stall =
+        nonmem_entered ? 0
+                       : nonmem_write_thread_.GetBegunCountOfOutstandingStall();
+    WaitForPendingWrites();
+
+    // A queue that was not entered must still be in the same stall, so that
+    // no writer in it can have allocated since.
+    const bool no_writer_in_flight =
+        (entered ||
+         (stall != 0 &&
+          write_thread_.GetBegunCountOfOutstandingStall() == stall)) &&
+        (nonmem_entered ||
+         (nonmem_stall != 0 &&
+          nonmem_write_thread_.GetBegunCountOfOutstandingStall() ==
+              nonmem_stall));
+    if (no_writer_in_flight) {
+      versions_->SyncLastSequenceWithAllocated();
+    }
+    if (nonmem_entered) {
+      nonmem_write_thread_.ExitUnbatched(&nonmem_w);
+    }
+    if (entered) {
+      write_thread_.ExitUnbatched(&w);
+    }
+    if (no_writer_in_flight) {
+      return;
+    }
+  }
+}
+
 // This function implements both manual and automatic background-error
 // recovery. It discards stale queued flush requests, waits for scheduled work
 // to exit, rebuilds and runs the flush work required by the recovery context,
@@ -425,16 +466,7 @@ Status DBImpl::ResumeImpl(DBRecoverContext context,
   // memtable-switch fence, after FlushAllColumnFamilies() has dropped and
   // re-acquired the DB mutex.
   if (two_write_queues_) {
-    WriteThread::Writer w;
-    write_thread_.EnterUnbatched(&w, &mutex_);
-    WriteThread::Writer nonmem_w;
-    nonmem_write_thread_.EnterUnbatched(&nonmem_w, &mutex_);
-    WaitForPendingWrites();
-
-    versions_->SyncLastSequenceWithAllocated();
-
-    nonmem_write_thread_.ExitUnbatched(&nonmem_w);
-    write_thread_.ExitUnbatched(&w);
+    SyncLastSequenceWithAllocatedAtRecoveryStart();
   }
 
   TEST_SYNC_POINT("DBImpl::ResumeImpl:AfterSyncSeq");
@@ -8037,6 +8069,7 @@ class FileIngestionHandleImpl : public FileIngestionHandle {
   FileIngestionHandleImpl& operator=(FileIngestionHandleImpl&&) = delete;
 
   Status Abort() override;
+  std::vector<SstFileMetaData> GetPreparedFiles() const override;
 
   DBImpl* const db_;
   const uint64_t first_file_number_;
@@ -8083,6 +8116,44 @@ Status FileIngestionHandleImpl::Abort() {
   return Status::OK();
 }
 
+std::vector<SstFileMetaData> FileIngestionHandleImpl::GetPreparedFiles() const {
+  std::vector<SstFileMetaData> files;
+  if (consumed_) {
+    return files;
+  }
+  for (const auto& job : jobs_) {
+    const auto& cf_paths = job->GetColumnFamilyData()->ioptions().cf_paths;
+    // Without allow_db_generated_files, commit assigns each file its sequence
+    // numbers, so they are unknown here.
+    const bool seqnos_known =
+        job->GetIngestionOptions().allow_db_generated_files;
+    for (const auto& file : job->files_to_ingest()) {
+      SstFileMetaData meta;
+      meta.directory = file.fd.GetPathId() < cf_paths.size()
+                           ? cf_paths[file.fd.GetPathId()].path
+                           : cf_paths.back().path;
+      meta.db_path = meta.directory;
+      meta.name = MakeTableFileName("", file.fd.GetNumber());
+      meta.relative_filename = meta.name.substr(1);
+      if (seqnos_known) {
+        meta.smallest_seqno = file.smallest_seqno;
+        meta.largest_seqno = file.largest_seqno;
+      }
+      meta.file_number = file.fd.GetNumber();
+      meta.size = file.fd.GetFileSize();
+      meta.temperature = file.file_temperature;
+      meta.file_checksum = file.file_checksum;
+      meta.file_checksum_func_name = file.file_checksum_func_name;
+      meta.smallestkey = file.smallest_internal_key.user_key().ToString();
+      meta.largestkey = file.largest_internal_key.user_key().ToString();
+      meta.num_entries = file.num_entries;
+      meta.num_deletions = file.table_properties.num_deletions;
+      files.push_back(std::move(meta));
+    }
+  }
+  return files;
+}
+
 Status DBImpl::PrepareFileIngestion(
     const std::vector<IngestExternalFileArg>& args,
     std::unique_ptr<FileIngestionHandle>* handle) {

db/db_impl/db_impl.h

@@ -2900,6 +2900,12 @@ class DBImpl : public DB
   // fence that is about to switch memtables.
   void MaybeSyncLastSequenceWithAllocatedForRecovery(FlushReason flush_reason);
 
+  // Like the memtable-switch fence, but does not wait for a write queue whose
+  // leader is in a write stall: that leader has not allocated sequence numbers
+  // yet, and it can be waiting for this recovery.
+  // REQUIRES: mutex locked, two_write_queues
+  void SyncLastSequenceWithAllocatedAtRecoveryStart();
+
   Status RetryFlushesForErrorRecovery(FlushReason flush_reason, bool wait);
 
   // Wait until flushing this column family won't stall writes

db/db_impl/db_impl_write.cc

@@ -2945,9 +2945,12 @@ Status DBImpl::DelayWrite(uint64_t num_bytes, WriteThread& write_thread,
     }
 
     // Don't wait if there's a background error that is not pending recovery
-    // since recovery might never be attempted.
+    // since recovery might never be attempted. Don't wait once the DB is
+    // stopped either: its recovery flush needs the write queue this writer
+    // holds.
     while ((error_handler_.GetBGError().ok() ||
-            error_handler_.IsRecoveryInProgress()) &&
+            (error_handler_.IsRecoveryInProgress() &&
+             !error_handler_.IsDBStopped())) &&
            write_controller_.IsStopped() &&
            !shutting_down_.load(std::memory_order_relaxed)) {
       if (write_options.no_slowdown) {

db/db_range_del_test.cc

@@ -1817,6 +1817,148 @@ TEST_F(DBRangeDelTest, CompactionEmitRangeTombstoneToSSTPartitioner) {
   ASSERT_EQ(2, NumTableFilesAtLevel(1));
 }
 
+namespace {
+// Records the compaction output files that were started, and how many of them
+// were empty and not kept.
+class CompactionOutputListener : public EventListener {
+ public:
+  void OnTableFileCreationStarted(
+      const TableFileCreationBriefInfo& info) override {
+    if (info.reason == TableFileCreationReason::kCompaction) {
+      std::lock_guard<std::mutex> lock(mutex_);
+      started_paths_.push_back(info.file_path);
+    }
+  }
+
+  void OnTableFileCreated(const TableFileCreationInfo& info) override {
+    if (info.reason == TableFileCreationReason::kCompaction &&
+        info.status.IsAborted()) {
+      std::lock_guard<std::mutex> lock(mutex_);
+      ++num_not_kept_;
+    }
+  }
+
+  std::vector<std::string> TakeStartedPaths() {
+    std::lock_guard<std::mutex> lock(mutex_);
+    return std::exchange(started_paths_, {});
+  }
+
+  int TakeNumNotKept() {
+    std::lock_guard<std::mutex> lock(mutex_);
+    return std::exchange(num_not_kept_, 0);
+  }
+
+ private:
+  std::mutex mutex_;
+  std::vector<std::string> started_paths_;
+  int num_not_kept_ = 0;
+};
+}  // namespace
+
+TEST_F(DBRangeDelTest, BottommostCompactionEmitRangeTombstoneToSSTPartitioner) {
+  // A range tombstone after the last point key must reach the SSTPartitioner
+  // at the bottommost level too. Otherwise the tombstone is appended to the
+  // file holding the point key and extends that file across the partition
+  // boundary, so no file starting at the boundary fits in the level.
+  // The fixed-prefix factory exercises the partitioner's prefix fast path,
+  // and SingleKeySstPartitionerFactory its per-key ShouldPartition() path.
+  // Each Key(i) is 9 bytes, so a 9-byte prefix puts every key in its own
+  // partition.
+  for (const auto& factory :
+       std::vector<std::shared_ptr<SstPartitionerFactory>>{
+           NewSstPartitionerFixedPrefixFactory(9),
+           std::make_shared<SingleKeySstPartitionerFactory>()}) {
+    SCOPED_TRACE(factory->Name());
+    Options options = CurrentOptions();
+    options.sst_partitioner_factory = factory;
+    options.disable_auto_compactions = true;
+    options.level_compaction_dynamic_level_bytes = true;
+    auto listener = std::make_shared<CompactionOutputListener>();
+    options.listeners.push_back(listener);
+    DestroyAndReopen(options);
+    const int bottommost_level = options.num_levels - 1;
+
+    ASSERT_OK(Put(Key(1), "val"));
+    // Keeps the tombstone from being dropped at the bottommost level
+    const Snapshot* snapshot = db_->GetSnapshot();
+    ASSERT_OK(db_->DeleteRange(WriteOptions(), db_->DefaultColumnFamily(),
+                               Key(2), Key(5)));
+    ASSERT_OK(Flush());
+
+    CompactRangeOptions cro;
+    cro.bottommost_level_compaction = BottommostLevelCompaction::kForce;
+    ASSERT_OK(db_->CompactRange(cro, nullptr, nullptr));
+
+    ColumnFamilyMetaData cf_meta;
+    db_->GetColumnFamilyMetaData(&cf_meta);
+    const std::vector<SstFileMetaData>& files =
+        cf_meta.levels[bottommost_level].files;
+    ASSERT_EQ(2, files.size());
+    ASSERT_EQ(Key(1), files[0].largestkey);
+    ASSERT_EQ(Key(2), files[1].smallestkey);
+    ASSERT_EQ(0, listener->TakeNumNotKept());
+    listener->TakeStartedPaths();
+
+    // Once nothing pins it, the tombstone is dropped. The file cut for it is
+    // still created, but it is empty, so the compaction deletes it before
+    // installing its outputs: it is never part of the LSM.
+    db_->ReleaseSnapshot(snapshot);
+    ASSERT_OK(db_->CompactRange(cro, nullptr, nullptr));
+    ASSERT_EQ(1, NumTableFilesAtLevel(bottommost_level));
+    ASSERT_EQ(1, listener->TakeNumNotKept());
+    const std::vector<std::string> started = listener->TakeStartedPaths();
+    ASSERT_EQ(2, started.size());
+    int num_on_disk = 0;
+    for (const auto& path : started) {
+      const Status exists = env_->FileExists(path);
+      if (exists.ok()) {
+        ++num_on_disk;
+      } else {
+        ASSERT_TRUE(exists.IsNotFound()) << exists.ToString();
+      }
+    }
+    ASSERT_EQ(1, num_on_disk);
+  }
+}
+
+TEST_F(DBRangeDelTest, BottommostRangeTombstoneAtSameUserKeyAsPointKey) {
+  // At the bottommost level a point key's seqno can be zeroed, so it sorts
+  // after a range tombstone that starts at the same user key. The partitioner
+  // must not cut between them, even one that requires a cut for every key.
+  Options options = CurrentOptions();
+  options.sst_partitioner_factory =
+      std::make_shared<SingleKeySstPartitionerFactory>();
+  options.disable_auto_compactions = true;
+  options.level_compaction_dynamic_level_bytes = true;
+  CompactRangeOptions cro;
+  cro.bottommost_level_compaction = BottommostLevelCompaction::kForce;
+
+  // The tombstone is dropped.
+  DestroyAndReopen(options);
+  ASSERT_OK(Put(Key(1), "val"));
+  ASSERT_OK(db_->DeleteRange(WriteOptions(), db_->DefaultColumnFamily(), Key(2),
+                             Key(5)));
+  ASSERT_OK(Put(Key(2), "val"));
+  ASSERT_OK(Flush());
+  ASSERT_OK(db_->CompactRange(cro, nullptr, nullptr));
+  ASSERT_EQ("val", Get(Key(2)));
+
+  // A newer tombstone covering the same user key is kept by a snapshot, so it
+  // is written to every output file that overlaps it.
+  DestroyAndReopen(options);
+  ASSERT_OK(db_->DeleteRange(WriteOptions(), db_->DefaultColumnFamily(), Key(2),
+                             Key(5)));
+  ASSERT_OK(Put(Key(2), "val"));
+  ASSERT_OK(Flush());
+  const Snapshot* snapshot = db_->GetSnapshot();
+  ASSERT_OK(db_->DeleteRange(WriteOptions(), db_->DefaultColumnFamily(), Key(1),
+                             Key(4)));
+  ASSERT_OK(db_->CompactRange(cro, nullptr, nullptr));
+  ASSERT_EQ("NOT_FOUND", Get(Key(2)));
+  ASSERT_EQ("val", Get(Key(2), snapshot));
+  db_->ReleaseSnapshot(snapshot);
+}
+
 TEST_F(DBRangeDelTest, OversizeCompactionGapBetweenPointKeyAndTombstone) {
   // L2 has 2 files
   // L2_0: 0, 1, 2, 3, 4

db/error_handler_fs_test.cc

@@ -413,6 +413,113 @@ TEST_F(DBErrorHandlingFSTest, FlushWriteFileScopeError) {
   }
 }
 
+// A retryable SST write error during a flush is a soft error that recovery
+// fixes by retrying the flush. A writer that waits for that flush must finish
+// once recovery is done.
+TEST_F(DBErrorHandlingFSTest, RetryableFlushErrorWithWriterStalledOnFlush) {
+  for (bool two_write_queues : {false, true}) {
+    SCOPED_TRACE("two_write_queues=" + std::to_string(two_write_queues));
+    Options options = GetDefaultOptions();
+    options.env = fault_env_.get();
+    options.create_if_missing = true;
+    options.two_write_queues = two_write_queues;
+    options.write_buffer_size = 64 << 10;
+    options.max_write_buffer_number = 2;
+    options.min_write_buffer_number_to_merge = 1;
+    options.bgerror_resume_retry_interval = 100000;
+    DestroyAndReopen(options);
+
+    IOStatus error_msg = IOStatus::IOError("Retryable IO Error");
+    error_msg.SetRetryable(true);
+    std::atomic<bool> injected{false};
+    // The first flush fails only after the writer stops on its memtables.
+    SyncPoint::GetInstance()->LoadDependency(
+        {{"DBImpl::DelayWrite:Wait", "BuildTable:BeforeFinishBuildTable"}});
+    SyncPoint::GetInstance()->SetCallBack(
+        "BuildTable:BeforeFinishBuildTable", [&](void*) {
+          if (!injected.exchange(true)) {
+            fault_fs_->SetFilesystemActive(false, error_msg);
+          }
+        });
+    SyncPoint::GetInstance()->SetCallBack(
+        "RecoverFromRetryableBGIOError:BeforeStart",
+        [&](void*) { fault_fs_->SetFilesystemActive(true); });
+    SyncPoint::GetInstance()->EnableProcessing();
+
+    const std::string value(40 << 10, 'v');
+    for (int i = 0; i < 8; ++i) {
+      ASSERT_OK(Put(Key(i), value));
+    }
+    ASSERT_TRUE(injected);
+    SyncPoint::GetInstance()->DisableProcessing();
+    SyncPoint::GetInstance()->ClearAllCallBacks();
+    SyncPoint::GetInstance()->LoadDependency({});
+    for (int i = 0; i < 8; ++i) {
+      ASSERT_EQ(value, Get(Key(i)));
+    }
+  }
+}
+
+// A retryable error syncing closed WALs during a flush is a hard error. A
+// writer that waits for that flush gets the error, and recovery then flushes
+// and resumes writes.
+TEST_F(DBErrorHandlingFSTest, FlushWALSyncErrorWithWriterStalledOnFlush) {
+  Options options = GetDefaultOptions();
+  options.env = fault_env_.get();
+  options.create_if_missing = true;
+  options.write_buffer_size = 64 << 10;
+  options.max_write_buffer_number = 2;
+  options.min_write_buffer_number_to_merge = 1;
+  options.bgerror_resume_retry_interval = 100000;
+  // With two column families, a flush syncs the closed WALs.
+  CreateAndReopenWithCF({"pikachu"}, options);
+
+  IOStatus error_msg = IOStatus::IOError("Retryable IO Error");
+  error_msg.SetRetryable(true);
+  std::atomic<bool> injected{false};
+  // The first flush fails only after the writer stops on its memtables.
+  SyncPoint::GetInstance()->LoadDependency(
+      {{"DBImpl::DelayWrite:Wait", "DBImpl::SyncClosedWals:Start"},
+       {"RecoverFromRetryableBGIOError:RecoverSuccess",
+        "DBErrorHandlingFSTest::FlushWALSyncErrorWithWriterStalledOnFlush:"
+        "Recovered"}});
+  SyncPoint::GetInstance()->SetCallBack(
+      "DBImpl::SyncClosedWals:Start", [&](void*) {
+        if (!injected.exchange(true)) {
+          fault_fs_->SetFilesystemActive(false, error_msg);
+        }
+      });
+  SyncPoint::GetInstance()->SetCallBack(
+      "RecoverFromRetryableBGIOError:BeforeStart",
+      [&](void*) { fault_fs_->SetFilesystemActive(true); });
+  SyncPoint::GetInstance()->EnableProcessing();
+
+  const std::string value(40 << 10, 'v');
+  Status s;
+  int num_written = 0;
+  while (num_written < 8) {
+    s = Put(Key(num_written), value);
+    if (!s.ok()) {
+      break;
+    }
+    ++num_written;
+  }
+  ASSERT_TRUE(injected);
+  ASSERT_TRUE(s.IsIOError()) << s.ToString();
+  ASSERT_EQ(s.severity(), Status::Severity::kHardError);
+
+  TEST_SYNC_POINT(
+      "DBErrorHandlingFSTest::FlushWALSyncErrorWithWriterStalledOnFlush:"
+      "Recovered");
+  SyncPoint::GetInstance()->DisableProcessing();
+  SyncPoint::GetInstance()->ClearAllCallBacks();
+  SyncPoint::GetInstance()->LoadDependency({});
+  ASSERT_OK(Put(Key(num_written), value));
+  for (int i = 0; i <= num_written; ++i) {
+    ASSERT_EQ(value, Get(Key(i)));
+  }
+}
+
 TEST_F(DBErrorHandlingFSTest, FlushWALWriteRetryableError) {
   std::shared_ptr<ErrorHandlerFSListener> listener =
       std::make_shared<ErrorHandlerFSListener>();

db/external_sst_file_ingestion_job.h

@@ -297,6 +297,10 @@ class ExternalSstFileIngestionJob {
 
   ColumnFamilyData* GetColumnFamilyData() const { return cfd_; }
 
+  const IngestExternalFileOptions& GetIngestionOptions() const {
+    return ingestion_options_;
+  }
+
   // Prepare the job by copying external files into the DB.
   Status Prepare(const std::vector<std::string>& external_files_paths,
                  const std::vector<std::string>& files_checksums,

db/external_sst_file_test.cc

@@ -8,6 +8,7 @@
 #include <atomic>
 #include <functional>
 #include <memory>
+#include <set>
 #include <sstream>
 
 #include "db/db_test_util.h"
@@ -846,6 +847,189 @@ TEST_F(ExternalSSTFileTest, AbortPreparedIngestion) {
   ASSERT_EQ("1", Get("x"));
 }
 
+static std::string DescribeFile(const SstFileMetaData& meta) {
+  return meta.directory + "/" + meta.relative_filename + " #" +
+         std::to_string(meta.file_number) + " " + std::to_string(meta.size) +
+         "B " + meta.file_checksum_func_name + ":" +
+         Slice(meta.file_checksum).ToString(/*hex=*/true) + " [" +
+         meta.smallestkey + ", " + meta.largestkey + "] temperature " +
+         std::to_string(static_cast<int>(meta.temperature));
+}
+
+static std::set<std::string> DescribeFiles(
+    const std::vector<SstFileMetaData>& files) {
+  std::set<std::string> described;
+  for (const auto& meta : files) {
+    described.insert(DescribeFile(meta));
+  }
+  return described;
+}
+
+static std::set<std::string> DescribeLiveFiles(DB* db) {
+  std::vector<LiveFileMetaData> live;
+  db->GetLiveFilesMetaData(&live);
+  return DescribeFiles({live.begin(), live.end()});
+}
+
+TEST_F(ExternalSSTFileTest, PreparedFilesAreTheFilesThatGoLive) {
+  Options options = CurrentOptions();
+  options.file_checksum_gen_factory = GetFileChecksumGenCrc32cFactory();
+  DestroyAndReopen(options);
+
+  std::string file_a;
+  ASSERT_OK(
+      GenerateExternalFileOnly(options, {{"a1", "1"}, {"a2", "2"}}, &file_a));
+  const std::string file_b = sst_files_dir_ + "put_and_delete.sst";
+  SstFileWriter writer(EnvOptions(), options);
+  ASSERT_OK(writer.Open(file_b));
+  ASSERT_OK(writer.Put("b", "2"));
+  ASSERT_OK(writer.Delete("c"));
+  ASSERT_OK(writer.Finish());
+  IngestExternalFileArg arg;
+  arg.column_family = db_->DefaultColumnFamily();
+  arg.external_files = {file_a, file_b};
+
+  std::unique_ptr<FileIngestionHandle> handle;
+  ASSERT_OK(db_->PrepareFileIngestion({arg}, &handle));
+  const std::vector<SstFileMetaData> prepared = handle->GetPreparedFiles();
+
+  ASSERT_EQ(2, prepared.size());
+  for (const auto& meta : prepared) {
+    const std::string path = meta.directory + "/" + meta.relative_filename;
+    uint64_t size_on_disk = 0;
+    ASSERT_OK(env_->GetFileSize(path, &size_on_disk));
+    ASSERT_EQ(size_on_disk, meta.size);
+    ASSERT_EQ("FileChecksumCrc32c", meta.file_checksum_func_name);
+    ASSERT_FALSE(meta.file_checksum.empty());
+  }
+  std::set<std::string> key_ranges;
+  for (const auto& meta : prepared) {
+    key_ranges.insert(meta.smallestkey + ".." + meta.largestkey + " " +
+                      std::to_string(meta.num_entries) + " entries " +
+                      std::to_string(meta.num_deletions) + " deletions");
+  }
+  ASSERT_EQ((std::set<std::string>{"a1..a2 2 entries 0 deletions",
+                                   "b..c 2 entries 1 deletions"}),
+            key_ranges);
+  ASSERT_EQ("NOT_FOUND", Get("a1"));
+
+  ASSERT_OK(db_->CommitFileIngestionHandle(std::move(handle)));
+
+  ASSERT_EQ(DescribeFiles(prepared), DescribeLiveFiles(db_.get()));
+}
+
+TEST_F(ExternalSSTFileTest, PreparedDbGeneratedFilesCarryTheirSeqnos) {
+  Options options = CurrentOptions();
+  DestroyAndReopen(options);
+
+  const std::string source_dbname = dbname_ + "_source";
+  ASSERT_OK(DestroyDB(source_dbname, options));
+  std::unique_ptr<DB> source_db;
+  ASSERT_OK(DB::Open(options, source_dbname, &source_db));
+  ASSERT_OK(source_db->Put(WriteOptions(), Key(1), "v1"));
+  ASSERT_OK(source_db->Put(WriteOptions(), Key(2), "v2"));
+  ASSERT_OK(source_db->Flush(FlushOptions()));
+  std::vector<LiveFileMetaData> source_meta;
+  source_db->GetLiveFilesMetaData(&source_meta);
+  ASSERT_EQ(1, source_meta.size());
+  ASSERT_LT(source_meta[0].smallest_seqno, source_meta[0].largest_seqno);
+
+  IngestExternalFileArg arg;
+  arg.column_family = db_->DefaultColumnFamily();
+  arg.external_files = {source_meta[0].directory + "/" +
+                        source_meta[0].relative_filename};
+  arg.options.allow_db_generated_files = true;
+  arg.options.snapshot_consistency = false;
+
+  std::unique_ptr<FileIngestionHandle> handle;
+  ASSERT_OK(db_->PrepareFileIngestion({arg}, &handle));
+  const std::vector<SstFileMetaData> prepared = handle->GetPreparedFiles();
+  ASSERT_EQ(1, prepared.size());
+  ASSERT_EQ(source_meta[0].smallest_seqno, prepared[0].smallest_seqno);
+  ASSERT_EQ(source_meta[0].largest_seqno, prepared[0].largest_seqno);
+
+  ASSERT_OK(db_->CommitFileIngestionHandle(std::move(handle)));
+
+  std::vector<LiveFileMetaData> live;
+  db_->GetLiveFilesMetaData(&live);
+  ASSERT_EQ(1, live.size());
+  ASSERT_EQ(live[0].smallest_seqno, prepared[0].smallest_seqno);
+  ASSERT_EQ(live[0].largest_seqno, prepared[0].largest_seqno);
+  ASSERT_EQ(live[0].name, prepared[0].name);
+  ASSERT_EQ(live[0].db_path, prepared[0].db_path);
+
+  ASSERT_OK(source_db->Close());
+  source_db.reset();
+  ASSERT_OK(DestroyDB(source_dbname, options));
+}
+
+TEST_F(ExternalSSTFileTest, PreparedFilesAreGoneAfterAbort) {
+  Options options = CurrentOptions();
+  DestroyAndReopen(options);
+
+  std::string file_path;
+  ASSERT_OK(GenerateExternalFileOnly(options, {{"x", "1"}}, &file_path));
+  IngestExternalFileArg arg;
+  arg.column_family = db_->DefaultColumnFamily();
+  arg.external_files = {file_path};
+
+  std::unique_ptr<FileIngestionHandle> handle;
+  ASSERT_OK(db_->PrepareFileIngestion({arg}, &handle));
+  const std::vector<SstFileMetaData> prepared = handle->GetPreparedFiles();
+  ASSERT_EQ(1, prepared.size());
+  const std::string path =
+      prepared[0].directory + "/" + prepared[0].relative_filename;
+  ASSERT_OK(env_->FileExists(path));
+
+  ASSERT_OK(handle->Abort());
+
+  ASSERT_TRUE(handle->GetPreparedFiles().empty());
+  ASSERT_TRUE(env_->FileExists(path).IsNotFound());
+}
+
+TEST_F(ExternalSSTFileTest, PreparedFilesOfMergedHandlesGoLive) {
+  Options options = CurrentOptions();
+  CreateAndReopenWithCF({"cf1", "cf2", "cf3"}, options);
+
+  std::string f_cf1;
+  std::string f_cf2_h1;
+  std::string f_cf2_h2;
+  std::string f_cf3;
+  ASSERT_OK(GenerateExternalFileOnly(options, {{"a", "1"}}, &f_cf1));
+  ASSERT_OK(GenerateExternalFileOnly(options, {{"b", "h1"}}, &f_cf2_h1));
+  ASSERT_OK(GenerateExternalFileOnly(options, {{"b", "h2"}}, &f_cf2_h2));
+  ASSERT_OK(GenerateExternalFileOnly(options, {{"c", "3"}}, &f_cf3));
+
+  // Each handle spans two column families, and both write cf2, so commit
+  // merges the two cf2 jobs.
+  std::vector<IngestExternalFileArg> args1(2);
+  args1[0].column_family = handles_[1];
+  args1[0].external_files = {f_cf1};
+  args1[1].column_family = handles_[2];
+  args1[1].external_files = {f_cf2_h1};
+  std::vector<IngestExternalFileArg> args2(2);
+  args2[0].column_family = handles_[2];
+  args2[0].external_files = {f_cf2_h2};
+  args2[1].column_family = handles_[3];
+  args2[1].external_files = {f_cf3};
+
+  std::unique_ptr<FileIngestionHandle> h1;
+  std::unique_ptr<FileIngestionHandle> h2;
+  ASSERT_OK(db_->PrepareFileIngestion(args1, &h1));
+  ASSERT_OK(db_->PrepareFileIngestion(args2, &h2));
+  std::set<std::string> prepared = DescribeFiles(h1->GetPreparedFiles());
+  prepared.merge(DescribeFiles(h2->GetPreparedFiles()));
+  ASSERT_EQ(4, prepared.size());
+
+  std::vector<std::unique_ptr<FileIngestionHandle>> batch;
+  batch.push_back(std::move(h1));
+  batch.push_back(std::move(h2));
+  ASSERT_OK(db_->CommitFileIngestionHandles(std::move(batch)));
+
+  ASSERT_EQ("h2", Get(2, "b"));
+  ASSERT_EQ(prepared, DescribeLiveFiles(db_.get()));
+}
+
 TEST_F(ExternalSSTFileTest, MultiHandleAtomicCommit) {
   Options options = CurrentOptions();
   CreateAndReopenWithCF({"cf1", "cf2", "cf3"}, options);

include/rocksdb/db.h

@@ -192,6 +192,37 @@ class FileIngestionHandle {
   // returns a Status.
   virtual Status Abort() = 0;
 
+  // Metadata of the ingested files this handle will make visible on commit.
+  // The files exist from prepare until commit or abort and keep their path,
+  // number, size and checksum when committed, so a caller can read them
+  // before they go live. Empty once the handle is spent.
+  //
+  // Set fields:
+  //   directory
+  //   relative_filename
+  //   name
+  //   db_path
+  //   file_number
+  //   size
+  //   temperature
+  //   file_checksum
+  //   file_checksum_func_name
+  //   smallestkey
+  //   largestkey
+  //   num_entries
+  //   num_deletions
+  //   smallest_seqno, largest_seqno: only with allow_db_generated_files;
+  //     otherwise commit assigns them
+  //
+  // Left at defaults (commit decides them):
+  //   epoch_number
+  //   file_creation_time
+  //   oldest_ancestor_time
+  //
+  // Does not include the range-tombstone file that an atomic_replace_range
+  // ingestion may add at commit.
+  virtual std::vector<SstFileMetaData> GetPreparedFiles() const = 0;
+
   FileIngestionHandle(const FileIngestionHandle&) = delete;
   FileIngestionHandle& operator=(const FileIngestionHandle&) = delete;
   FileIngestionHandle(FileIngestionHandle&&) = delete;

include/rocksdb/version.h

@@ -12,7 +12,7 @@
 // NOTE: in 'main' development branch, this should be the *next*
 // minor or major version number planned for release.
 #define ROCKSDB_MAJOR 11
-#define ROCKSDB_MINOR 14
+#define ROCKSDB_MINOR 15
 #define ROCKSDB_PATCH 0
 
 // Make it easy to do conditional compilation based on version checks, i.e.

table/table_test.cc

@@ -8680,11 +8680,15 @@ TEST_F(ExternalTableTest, FactoryWithoutTableBuilderIsNeverCompacted) {
     return names;
   };
   std::vector<std::string> ingested[3];
+  // Otherwise the FIFO compaction of handles_[1] can delete its file before
+  // the check below sees it.
+  ASSERT_OK(db_->PauseBackgroundWork());
   for (int cf : {1, 2}) {
     ASSERT_OK(db_->IngestExternalFile(handles_[cf], {file}, ingest_options));
     ingested[cf] = live_files(cf);
     ASSERT_EQ(ingested[cf].size(), 1);
   }
+  ASSERT_OK(db_->ContinueBackgroundWork());
   ASSERT_OK(dbfull()->TEST_WaitForCompact());
   ASSERT_TRUE(live_files(1).empty());
   ASSERT_EQ(live_files(2), ingested[2]);

tools/check_format_compatible.sh

@@ -161,7 +161,7 @@ EOF
 
 # To check for DB forward compatibility with loading options (old version
 # reading data from new), as well as backward compatibility
-declare -a db_forward_with_options_refs=("10.4.fb" "10.5.fb" "10.6.fb" "10.7.fb" "10.8.fb" "10.9.fb" "10.10.fb" "10.11.fb" "11.0.fb" "11.1.fb" "11.2.fb" "11.3.fb" "11.4.fb" "11.5.fb" "11.6.fb" "11.7.fb" "11.8.fb" "11.9.fb" "11.10.fb" "11.11.fb" "11.12.fb" "11.13.fb")
+declare -a db_forward_with_options_refs=("10.4.fb" "10.5.fb" "10.6.fb" "10.7.fb" "10.8.fb" "10.9.fb" "10.10.fb" "10.11.fb" "11.0.fb" "11.1.fb" "11.2.fb" "11.3.fb" "11.4.fb" "11.5.fb" "11.6.fb" "11.7.fb" "11.8.fb" "11.9.fb" "11.10.fb" "11.11.fb" "11.12.fb" "11.13.fb" "11.14.fb")
 # To check for DB forward compatibility without loading options (in addition
 # to the "with loading options" set), as well as backward compatibility
 declare -a db_forward_no_options_refs=() # N/A at the moment

unreleased_history/behavior_changes/async_io_direct_io_multiget.md

@@ -1 +0,0 @@
-With `ReadOptions::async_io`, `MultiGet` now reads data blocks of direct-IO files with asynchronous reads (`FSRandomAccessFile::ReadAsync`) instead of a synchronous `MultiRead`. Each block read is aligned separately rather than merged with adjacent reads. The bytes read are still charged to `DBOptions::rate_limiter` at `ReadOptions::rate_limiter_priority`.

unreleased_history/behavior_changes/async_read_rate_limiter.md

@@ -1 +0,0 @@
-Asynchronous file reads (`FSRandomAccessFile::ReadAsync`, used with `ReadOptions::async_io` by `MultiGet` and iterator prefetching) are now charged to `DBOptions::rate_limiter` at `ReadOptions::rate_limiter_priority`, like synchronous reads. A read is charged once the file system accepts it.

unreleased_history/behavior_changes/create_cf_validates_table_factory.md

@@ -1 +0,0 @@
-`CreateColumnFamily()` and `CreateColumnFamilies()` now run the table factory option validation that `DB::Open()` runs, so they return `InvalidArgument` instead of creating a column family that the next `DB::Open()` would reject. External table factories wrapped by `NewExternalTableFactory()` now receive `ValidateOptions()` calls.

unreleased_history/behavior_changes/fifo_temperature_copy_listener_events.md

@@ -1 +0,0 @@
-With `CompactionOptionsFIFO::allow_trivial_copy_when_change_temperature`, each file copied for a temperature change is now reported to `EventListener::OnTableFileCreationStarted()` and `OnTableFileCreated()` (with the original file's table properties) and included in `CompactionJobInfo::table_properties`.

unreleased_history/behavior_changes/get_merge_operands_reset_written_slots.md

@@ -1 +0,0 @@
-`DB::GetMergeOperands()` now resets only the `merge_operands` entries it fills, instead of all `expected_max_number_of_operands` entries. Entries past `*number_of_operands`, and the whole array on a non-OK result, are left as the caller passed them.

unreleased_history/behavior_changes/json_writer_escape.md

@@ -1 +0,0 @@
-The JSON written to the info LOG by `EventLogger` (`EVENT_LOG_v1` lines) and by `VersionEdit::DebugJSON()` now escapes special characters in keys and string values, and writes non-finite floating-point values as `null`, so the output is always valid JSON.

unreleased_history/behavior_changes/periodic_compaction_phase_catch_up.md

@@ -1 +0,0 @@
-Improved how `periodic_compaction_phase_recovery_percent` spreads catch-up periodic compactions after it is enabled or changed (or the DB is reopened). This reduces a transient burst in periodic compactions when many DBs make such a change at once.

unreleased_history/behavior_changes/remote_compaction_pool.md

@@ -1 +0,0 @@
-The new DB option `use_separate_remote_compaction_pool` defaults to false and can be changed through `SetDBOptions()` for newly scheduled jobs. When enabled with CompactionService, remote jobs use a dedicated pool and each local subcompaction uses LOW/BOTTOM and counts against the DB compaction limit. Existing jobs finish using their original scheduling model.

unreleased_history/behavior_changes/set_options_skip_unchanged.md

@@ -1 +0,0 @@
-`DB::SetOptions()` now returns OK without writing a new OPTIONS file or installing a new SuperVersion when the call changes no option value, as `DB::SetDBOptions()` already does. Calls that set an object option, such as `table_factory` or one of its fields, are still applied in full. After a failed OPTIONS file write, the next `SetOptions()` or `SetDBOptions()` call writes the file even if no option value changes.

unreleased_history/behavior_changes/udi_validation_and_cache_statistics.md

@@ -1,3 +0,0 @@
-Custom index factories are rejected with user-defined timestamps in every index mode except `kStandardOnly`, which ignores the factory. Custom-index block-cache hits, misses, and adds now contribute to `BLOCK_CACHE_INDEX_*` instead of `BLOCK_CACHE_DATA_*`.
-
-Trie index construction returns `NotSupported` when a data-block offset or size exceeds `UINT32_MAX`, including when the trie is optional in `kStandardDefault`. Flush, compaction, and `SstFileWriter::Finish()` can therefore fail for such output; an SST larger than 4 GiB is not itself the rejection condition. File-size targets do not split L0 outputs or `SstFileWriter` outputs. Bound those outputs or disable trie construction with `kStandardOnly`.

unreleased_history/bug_fixes/atomic_replace_range_compaction_race.md

@@ -1 +0,0 @@
-Fixed a race where `IngestExternalFiles()` with `atomic_replace_range` could delete an existing SST file that a concurrently picked compaction (e.g. a universal compaction outputting to L0) used as input, causing the compaction to fail with "Corruption: VersionBuilder: Cannot delete table file ... since it is not in the LSM tree" and putting the DB into read-only mode. Atomic range replacement now returns `NotSupported` with FIFO compaction.

unreleased_history/bug_fixes/blob_direct_write_failed_seal_rollback.md

@@ -1 +0,0 @@
-Fixed blob direct write when sealing a blob file fails during a flush or a write: the flush left its memtables marked as being flushed, so no later flush picked them up and error recovery waited for them forever, and the blob file was dropped from tracking, so a retried flush could install table files whose blob file was never added to the MANIFEST.

unreleased_history/bug_fixes/blob_direct_write_overlapping_flushes.md

@@ -1 +0,0 @@
-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.

unreleased_history/bug_fixes/blob_direct_write_space_limit_seal.md

@@ -1 +0,0 @@
-Fixed blob direct write when the `SstFileManager` space limit is reached while a blob file is sealed: the sealed blob file was deleted while memtables still referenced it, so reads of its values failed and the values were lost, and reporting the error could deadlock with a concurrent memtable switch. The seal and the write or flush now succeed and keep the file, and writes then stop with a `SpaceLimit` background error, so for blob direct write the limit is soft: usage can exceed it by up to about `blob_file_size` × `blob_direct_write_partitions` for each memtable not flushed yet. Also fixed a write batch that failed after writing some of its values to blob files: those blobs are now counted as garbage, so their blob file can still be deleted.

unreleased_history/bug_fixes/blob_timestamp_history.md

@@ -1 +0,0 @@
-Values that retain external blob references now keep their original timestamps during history compaction, preventing blob key verification failures.

unreleased_history/bug_fixes/bottommost_range_del_partitioner_cut.md

@@ -0,0 +1 @@
+Fixed a bug where a bottommost-level compaction could ignore an `SstPartitioner` boundary between the last point key of an output file and a following range tombstone kept alive by a snapshot. The tombstone extended that file across the boundary, so an `IngestExternalFile()` with `fail_if_not_bottommost_level=true` into the range starting at the boundary could fail with `TryAgain`.

unreleased_history/bug_fixes/clip_empty_user_key.md

@@ -1 +0,0 @@
-Fixed `DB::ClipColumnFamily()` leaving an empty-string key outside the kept range when it is the only key in an L0 SST file.

unreleased_history/bug_fixes/common_prefix_reverse_comparator.md

@@ -1 +0,0 @@
-Fixed `format_version=8` with `ReverseBytewiseComparator`: a block's stored common key prefix could extend past a user key (one following longer user keys that extend it), so reads and iteration could miss that key (or its deletion, exposing an older value), and compactions could fail with `Corruption`. Files written by affected versions (11.10 through 11.13) are now read correctly, and new files no longer have this problem.

unreleased_history/bug_fixes/error_recovery_behind_stalled_writer.md

@@ -0,0 +1 @@
+Fixed automatic error recovery hanging forever, with all writes stopped, when a writer was waiting on a write stall (for example the `max_write_buffer_number` memtable limit) as a flush failed. This happened for a retryable SST write error with `two_write_queues = true`, and for a hard error such as a failed WAL sync during the flush in any configuration. A writer that is waiting when the DB stops now returns the background error, like a new write does.

unreleased_history/bug_fixes/file_creation_time_always_set.md

@@ -1 +0,0 @@
-Fixed SST file times that could defer periodic compaction and TTL. Files written by `SstFileWriter`, flushed during WAL recovery, or recovered by `RepairDB` now record a file creation time. Files added by `IngestExternalFile()` or `CreateColumnFamilyWithImport()` now keep the creation and oldest ancestor times from the file or export metadata (capped at the current time) instead of taking the time they were added, so ingesting or importing old data can make it due for periodic compaction or TTL soon afterward. FIFO TTL compaction now also deletes ingested `SstFileWriter` files, which previously blocked TTL deletion of themselves and all newer files. Files copied by FIFO compaction for a temperature change now keep the original file's creation time.

unreleased_history/bug_fixes/lru_uncharged_standalone_get_charge.md

@@ -1 +0,0 @@
-Fixed a debug-build assertion failure and incorrect cache charge and block-cache byte statistics in release builds when a full, strict-capacity LRU cache with metadata charging creates an uncharged standalone handle, including during secondary-cache promotion.

unreleased_history/bug_fixes/manual_compaction_atomic_flush_cf_drop.md

@@ -1 +0,0 @@
-Fixed atomic flushes of all column families when another column family is dropped while they wait: `CompactRange()` no longer fails with a "Column family dropped" error, and `WaitForCompact()` and DB close no longer return OK without flushing the remaining column families.

unreleased_history/bug_fixes/mempurge_filter_error_relock.md

@@ -1 +0,0 @@
-Fixed a bug where MemPurge returned with the DB mutex unlocked when the compaction filter created for flush did not ignore snapshots.

unreleased_history/bug_fixes/multiscan_prepare_l0_crash.md

@@ -1 +0,0 @@
-Fixed a crash in MultiScan `Iterator::Prepare()` with a single scan range without a limit when the DB has L0 files, and an assertion failure in debug builds when `Prepare()` is called on an already-positioned iterator. `Prepare()` now leaves the iterator invalid until it is positioned again (e.g. by `Seek()`).

unreleased_history/bug_fixes/partitioned_filter_multiget_out_of_domain.md

@@ -1 +0,0 @@
-Fixed `MultiGet` returning NotFound for existing keys that are outside the prefix extractor's domain, with partitioned filters, `whole_key_filtering=false`, and filter partitions cut together with index partitions (`decouple_partitioned_filters=false`, also the layout of files written by older versions). A filter partition holding only such keys is empty, and `MultiGet` treated every key mapped to it as absent. `Get` was not affected.

unreleased_history/bug_fixes/prepared_ingestion_sst_retention.md

@@ -1 +0,0 @@
-Fixed prepared external file ingestions retaining unrelated obsolete SSTs until commit or abort. Prepared ingestions now protect only their staged files from deletion.

unreleased_history/bug_fixes/range_conversion_ingest_older_iterator.md

@@ -1 +0,0 @@
-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.

unreleased_history/bug_fixes/range_tombstone_conversion_memtable_tier.md

@@ -1 +0,0 @@
-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`.

unreleased_history/bug_fixes/recycled_wal_truncated_by_old_writer.md

@@ -1 +0,0 @@
-Fixed a race with `recycle_log_file_num > 0` where an obsolete WAL could be reused for a new WAL while its old writer was still open. Closing the old writer then truncated the reused file to the old WAL's size, so writes in the new WAL past that point, including acknowledged `sync` writes, were lost on the next reopen.

unreleased_history/bug_fixes/remote_compaction_tiering_options.md

@@ -1 +0,0 @@
-Fixed remote compaction ignoring `preclude_last_level_data_seconds` and `preserve_internal_time_seconds`, which caused recent data to be written to the last level and recent sequence numbers to be zeroed. Takes effect once both the primary and the compaction worker are upgraded.

unreleased_history/bug_fixes/resume_failure_subcompaction_completed.md

@@ -1 +0,0 @@
-Fixed `DB::OpenAndCompact()` with `OpenAndCompactOptions::allow_resumption` not calling `EventListener::OnSubcompactionCompleted()` when the compaction fails to resume from its saved progress.

unreleased_history/bug_fixes/secondary_manifest_switch_empty_cf.md

@@ -1 +0,0 @@
-Fixed MANIFEST catch-up on secondary instances (`TryCatchUpWithPrimary()`) and follower instances installing a column family with no files and still returning OK. This happened when catch-up switched to a new MANIFEST and the primary deleted a file that MANIFEST lists before catch-up checked it, recording the deletion in a newer MANIFEST. Catch-up now keeps the column family's previous view until a complete state from the new MANIFEST is available.

unreleased_history/bug_fixes/udi_cache_and_legacy_options.md

@@ -1,5 +0,0 @@
-Custom index blocks now reuse cached contents on SST reopen, and deprecated UDI boolean options can be cleared without overriding an explicitly configured index mode.
-
-Legacy UDI options copied from a factory can now be cleared before constructing a new factory without retaining the derived mode. The C API preserves both that rollback behavior and explicitly selected modes. OPTIONS files serialize equivalent legacy booleans for `kStandardDefault`, `kStandardRequired`, and `kCustomDefault` so older readers retain routing.
-
-Empty custom-index output retains the standard-index fallback in modes that write both indexes; custom-only output must remain nonempty. Default primary reads retain the table factory's precedence over a legacy per-read factory pointer, while explicit custom selectors still check its name.

unreleased_history/bug_fixes/udi_options_and_memory_compatibility.md

@@ -1,3 +0,0 @@
-Preserve legacy UDI routing when loading old OPTIONS files, and reject invalid index modes in standalone SST readers and writers. Table readers now charge the custom index blocks they own; a custom reader whose `ApproximateMemoryUsage()` already includes that block should override `IndexFactoryReader::MemoryUsageIncludesIndexBlock()` to return true so the block is not charged twice.
-
-Keep cached custom-index Slice views intact when a reader consumes its input. Register custom-index cache helpers without requiring a later change. Trie index size estimates now account for shared prefixes, preventing premature compaction file cuts, and reject data-block offsets or sizes exceeding the 32-bit on-disk limit instead of truncating them.

unreleased_history/new_features/async_read_cpu_time.md

@@ -1 +0,0 @@
-Added experimental `SetAsyncReadCPUTimeEnabled()`: the next async or coroutine read started on the thread records its CPU time in `PerfContext::get_cpu_nanos` at any perf level except `kDisable`, without turning on the other timers of `kEnableTimeAndCPUTimeExceptForMutex`.

unreleased_history/new_features/db_split_merge.md

@@ -1 +0,0 @@
-Added `SplitDB()`, `DeleteSplitRanges()`, and `MergeDB()` in `rocksdb/utilities/db_split_merge.h`, plus a `db_split_merge` command-line tool, for moving selected column-family key ranges into a new DB and copying ranges into existing column families.

unreleased_history/new_features/global_mutex_hold_logging.md

@@ -1 +0,0 @@
-Added opt-in sampled logging of long-held DB-wide mutexes under `StatsLevel::kAll`, with a runtime-changeable `mutex_hold_log_threshold_us` DB option defaulting to 0 (disabled).

unreleased_history/new_features/live_tune.md

@@ -1 +0,0 @@
-Add experimental live tuning of a DB's options (`StartLiveTuneCycle()`, `GetLiveTuneStatus()`, `StopLiveTuneCycle()` in `include/rocksdb/live_tune.h`): one call runs a cycle that measures the DB, asks an application-provided `LiveTuneAdvisor` for option changes, applies them, and reverts them when they break the application's constraints or do not prove better.

unreleased_history/new_features/merge_before_compaction_filter.md

@@ -1 +0,0 @@
-Added `CompactionFilter::GetMergeOrder()` and `CompactionFilterFactory::GetMergeOrder()` to optionally run the merge operator before the compaction filter during compaction, so that the filter sees merge results. The default behavior is unchanged.

unreleased_history/new_features/parallel_compression_with_udi.md

@@ -1 +0,0 @@
-Parallel compression (`CompressionOptions::parallel_threads > 1`) now works with a `user_defined_index_factory` whose builders support it, including the built-in trie index. Previously a custom index always forced single-threaded compression.

unreleased_history/new_features/prepared_files.md

@@ -0,0 +1 @@
+Added `FileIngestionHandle::GetPreparedFiles()`: the metadata (path, file number, size, checksum, key range, entry and deletion counts) of a prepared ingestion's files. The files keep their path, number, size and checksum when the handle is committed, so a caller can read the files before they become visible. The list does not include the range-tombstone file that an `atomic_replace_range` ingestion may add at commit.

unreleased_history/new_features/reject_too_old_file.md

@@ -1 +0,0 @@
-Added option `reject_too_old_file` to `IngestExternalFileOptions` and `ImportColumnFamilyOptions` to reject files whose data is already older than the column family's `periodic_compaction_seconds` (or `ttl` for FIFO compaction).

unreleased_history/new_features/remote_compaction_blob_files.md

@@ -1,7 +0,0 @@
-Added BlobDB support for remote compaction. Blob garbage collection uses the
-cutoff computed by the primary; workers receiving legacy input without a cutoff
-safely disable blob garbage collection for that job. When the primary rejects a
-remote blob result because its metadata, files, or capabilities fail validation,
-it reruns the compaction locally, so `CompactRange` and `CompactFiles` succeed.
-`CompactionService` `kFailure` semantics are unchanged; ZippyDB production also
-maps service failures to local fallback through `HandleFailure(FALLBACK_LOCAL)`.

unreleased_history/new_features/skip_missing_blob_files.md

@@ -1 +0,0 @@
-`CompactRangeOptions::skip_missing_blob_files` enables offline repair that deletes values referencing integrated external blob files absent from Version metadata, preserves available wide entity columns, and rewrites selected SSTs locally, bypassing `DBOptions::compaction_service`. It defaults to false; FIFO compaction and legacy BlobDB are unsupported.

unreleased_history/new_features/sst_file_db_ownership_metadata.md

@@ -1 +0,0 @@
-Added `db_ownership_start_time` and `data_file_provenance` (decoded by `DataFileProvenance`) to `SstFileMetaData`, recording when the DB began holding each SST file and how it got the file (e.g. flush, compaction, ingestion, or import).

unreleased_history/new_features/table_factory_without_builder.md

@@ -1 +0,0 @@
-Added `TableFactory::IsTableBuilderSupported()` and `ExternalTableFactory::IsTableBuilderSupported()` for table factories that only read ingested or imported files. A column family whose factory returns false must set `disallow_memtable_writes`, is never scheduled for automatic compaction, does not stall writes for L0 files or pending compaction bytes, and returns `NotSupported` from `CompactRange()` and `CompactFiles()` instead of failing a background compaction.

unreleased_history/new_features/trace_analyzer_put_entity.md

@@ -1 +0,0 @@
-trace_analyzer: new `--analyze_put_entity` flag analyzes PutEntity queries (previously their statistics were neither enabled nor disabled explicitly).

unreleased_history/new_features/trace_read_write_options.md

@@ -1 +0,0 @@
-Added `TraceOptions::include_read_write_options`: query traces can record the main `ReadOptions` and `WriteOptions` of each traced call (e.g. `fill_cache`, `disableWAL`), which replay then uses instead of default options.

unreleased_history/performance_improvements/pin_data_immortal_mmap_no_pin.md

@@ -1 +0,0 @@
-With `ReadOptions::pin_data`, iterators over read-only DBs opened with `max_open_files = -1` and `allow_mmap_reads = true` no longer accumulate memory as they move across SST files.

unreleased_history/performance_improvements/snapshot_without_db_mutex.md

@@ -1 +0,0 @@
-On a read-write primary, `DB::GetSnapshot()` and `DB::ReleaseSnapshot()` no longer acquire the DB mutex in the common case. Secondary and follower snapshot creation retains the DB mutex needed to synchronize with view installation, while release remains sharded. The snapshot list is now split into 16 per-CPU shards, each with its own lock, so primary threads creating and releasing snapshots no longer serialize on one lock or wait behind flush, compaction, or other work holding the DB mutex. Snapshot-release maintenance is deferred to existing flush, compaction, and periodic maintenance paths. Primary snapshot creation serializes with the DB-mutex-held phases of external file ingestion but can proceed while the MANIFEST write temporarily releases that mutex.

unreleased_history/public_api_changes/encode_entry_buffer.md

@@ -1 +0,0 @@
-Added `GetEncodedEntrySize()` and an `EncodeEntry()` overload that writes internal keys into caller-owned buffers without allocating on success.

unreleased_history/public_api_changes/env_read_thread_pool.md

@@ -1 +0,0 @@
-Added `Env::Priority::READ` and `ThreadStatus::READ_PRIORITY` for a dedicated Env-managed read-work thread pool. `Env::Priority::USER` remains reserved for internal statistics and is not backed by a thread pool.

unreleased_history/public_api_changes/experimental_multiscan_parallel_decompression.md

@@ -1 +0,0 @@
-Added experimental `MultiScanArgs::use_parallel_async_io_block_decompression` for materializing asynchronously prefetched data blocks, including decompression, in parallel on the `Env::Priority::READ` thread pool. A zero-sized pool preserves inline materialization.

unreleased_history/public_api_changes/index_factory.md

@@ -1 +0,0 @@
-Added the experimental `IndexFactory` interface (`include/rocksdb/index_factory.h`) for pluggable SST indexes; the `UserDefinedIndex*` names remain as aliases. `BlockBasedTableOptions::index_mode` chooses which indexes an SST is built with and which one reads use by default, superseding the now-deprecated `use_udi_as_primary_index` and `fail_if_no_udi_on_open`. `ReadOptions::read_index` overrides that choice per read, and the `standard_index_is_stub` table property marks SSTs built with `kCustomOnly`. Both options are also available in the C API.

unreleased_history/public_api_changes/multi_get_entity_with_metadata.md

@@ -1 +0,0 @@
-Added experimental `DB::GetEntityWithMetadata()` and `DB::MultiGetEntityWithMetadata()`, the wide-column entity counterparts of `DB::GetWithMetadata()` and `DB::MultiGetWithMetadata()` with opt-in explicit-snapshot newer-version detection. Requesting key timestamps returns `NotSupported`.

unreleased_history/public_api_changes/remove_solaris_support.md

@@ -1 +0,0 @@
-Removed support for building on Solaris, which had not compiled since 5.18. Removed RocksJava `Environment.isSolaris()`.

utilities/transactions/write_prepared_transaction_seqno_test.cc

@@ -238,6 +238,160 @@ TEST_F(WritePreparedTransactionSeqnoTest,
   Close();
 }
 
+// Recovery does not wait for a main write queue whose leader is in a write
+// stall. If that stall ends while recovery is joining the nonmem queue, the
+// leader can allocate a sequence range before recovery syncs. Recovery must
+// then retry and wait for the leader to publish, instead of publishing the
+// leader's allocation for it.
+TEST_F(WritePreparedTransactionSeqnoTest,
+       RecoverySeqnoSyncRetriesWhenStallEndsDuringNonmemJoin) {
+  txn_db_options_.write_policy = TxnDBWritePolicy::WRITE_UNPREPARED;
+  ASSERT_OK(Open());
+  TransactionDB& db = AssertNotNull(db_);
+  DBImpl& db_impl =
+      AssertNotNull(static_cast_with_check<DBImpl>(db.GetRootDB()));
+  VersionSet& versions = AssertNotNull(db_impl.GetVersionSet());
+  SyncPoint& sync_point = AssertNotNull(SyncPoint::GetInstance());
+
+  const std::string test_prefix =
+      "RecoverySeqnoSyncRetriesWhenStallEndsDuringNonmemJoin";
+  sync_point.LoadDependency(
+      {{"DBImpl::DelayWrite:Wait", test_prefix + ":writer_stalled"},
+       {"DBImpl::WriteImpl:AfterWBWIIngestBeforeSetLastSequence:pause",
+        test_prefix + ":writer_paused"},
+       {test_prefix + ":release_writer",
+        "DBImpl::WriteImpl:AfterWBWIIngestBeforeSetLastSequence:resume"}});
+
+  std::unique_ptr<WriteControllerToken> stop_token =
+      db_impl.TEST_write_controler().GetStopToken();
+  std::atomic<int> non_blocking_joins{0};
+  std::atomic<bool> release_writer_called{false};
+  std::atomic<bool> recovery_waited_for_writer{false};
+  std::atomic<bool> writer_published_before_sync{false};
+  auto release_writer = [&]() {
+    if (!release_writer_called.exchange(true)) {
+      sync_point.Process(test_prefix + ":release_writer");
+    }
+  };
+  // The first non-blocking join is refused by the main queue stall. The second
+  // one joins the nonmem queue without the DB mutex. End the stall there and
+  // let the stalled writer allocate and pause before it publishes.
+  sync_point.SetCallBack("WriteThread::EnterUnbatchedNonBlocking:BeforeLock",
+                         [&](void*) {
+                           if (non_blocking_joins.fetch_add(1) == 1) {
+                             stop_token.reset();
+                             db_impl.TEST_SignalAllBgCv();
+                             sync_point.Process(test_prefix + ":writer_paused");
+                           }
+                         });
+  // On retry, recovery queues behind the paused writer in the main queue.
+  sync_point.SetCallBack("WriteThread::EnterUnbatched:Wait", [&](void*) {
+    recovery_waited_for_writer.store(true);
+    release_writer();
+  });
+  // Fallback so a path that syncs without retrying fails instead of hanging.
+  sync_point.SetCallBack("DBImpl::ResumeImpl:AfterSyncSeq", [&](void*) {
+    writer_published_before_sync.store(release_writer_called.load());
+    release_writer();
+  });
+  sync_point.EnableProcessing();
+
+  WriteOptions write_opts;
+  TransactionOptions txn_opts;
+  Status prepare_status;
+  std::unique_ptr<Transaction> stalled_txn(
+      db.BeginTransaction(write_opts, txn_opts));
+  Transaction& stalled_txn_ref = AssertNotNull(stalled_txn.get());
+  ASSERT_OK(stalled_txn_ref.SetName("stalled_txn"));
+  ASSERT_OK(stalled_txn_ref.Put("stalled_key", "stalled_value"));
+  std::thread writer([&]() { prepare_status = stalled_txn_ref.Prepare(); });
+
+  // The writer leads the main queue and waits on the write stop.
+  TEST_SYNC_POINT(test_prefix + ":writer_stalled");
+  DBRecoverContext context(FlushReason::kErrorRecoveryRetryFlush);
+  Status resume_status = db_impl.TEST_ResumeImpl(context);
+  writer.join();
+  sync_point.DisableProcessing();
+  sync_point.ClearAllCallBacks();
+  sync_point.LoadDependency({});
+
+  ASSERT_OK(resume_status);
+  ASSERT_OK(prepare_status);
+  ASSERT_GE(non_blocking_joins.load(), 3);
+  ASSERT_TRUE(recovery_waited_for_writer.load());
+  ASSERT_TRUE(writer_published_before_sync.load());
+  ASSERT_EQ(versions.LastSequence(), versions.LastAllocatedSequence());
+  ASSERT_OK(stalled_txn_ref.Commit());
+
+  stalled_txn.reset();
+  Close();
+}
+
+// Recovery does not wait for a nonmem write queue whose leader, a WAL-only
+// commit, is in a write stall. The stalled commit has not allocated, so
+// recovery syncs and finishes while the commit is still waiting.
+TEST_F(WritePreparedTransactionSeqnoTest,
+       RecoverySeqnoSyncSkipsStalledWalOnlyCommit) {
+  txn_db_options_.write_policy = TxnDBWritePolicy::WRITE_UNPREPARED;
+  ASSERT_OK(Open());
+  TransactionDB& db = AssertNotNull(db_);
+  DBImpl& db_impl =
+      AssertNotNull(static_cast_with_check<DBImpl>(db.GetRootDB()));
+  VersionSet& versions = AssertNotNull(db_impl.GetVersionSet());
+  SyncPoint& sync_point = AssertNotNull(SyncPoint::GetInstance());
+
+  WriteOptions write_opts;
+  TransactionOptions txn_opts;
+  // The data goes to the memtable at Prepare(), so Commit() writes only a WAL
+  // marker on the nonmem write queue.
+  std::unique_ptr<Transaction> wal_only_commit_txn(
+      db.BeginTransaction(write_opts, txn_opts));
+  Transaction& wal_only_commit_txn_ref =
+      AssertNotNull(wal_only_commit_txn.get());
+  ASSERT_OK(wal_only_commit_txn_ref.SetName("wal_only_commit_txn"));
+  ASSERT_OK(wal_only_commit_txn_ref.Put("commit_key", "commit_value"));
+  ASSERT_OK(wal_only_commit_txn_ref.Prepare());
+
+  const std::string test_prefix = "RecoverySeqnoSyncSkipsStalledWalOnlyCommit";
+  sync_point.LoadDependency(
+      {{"DBImpl::DelayWrite:NonmemWait", test_prefix + ":commit_stalled"}});
+  sync_point.EnableProcessing();
+
+  std::unique_ptr<WriteControllerToken> stop_token =
+      db_impl.TEST_write_controler().GetStopToken();
+  std::atomic<bool> commit_done{false};
+  Status commit_status;
+  std::thread committer([&]() {
+    commit_status = wal_only_commit_txn_ref.Commit();
+    commit_done.store(true);
+  });
+
+  // The commit leads the nonmem queue and waits on the write stop.
+  TEST_SYNC_POINT(test_prefix + ":commit_stalled");
+  DBRecoverContext context(FlushReason::kErrorRecoveryRetryFlush);
+  Status resume_status = db_impl.TEST_ResumeImpl(context);
+  const bool commit_done_at_recovery_end = commit_done.load();
+  const bool synced =
+      versions.LastSequence() == versions.LastAllocatedSequence();
+
+  stop_token.reset();
+  db_impl.TEST_SignalAllBgCv();
+  committer.join();
+  sync_point.DisableProcessing();
+  sync_point.LoadDependency({});
+
+  ASSERT_OK(resume_status);
+  ASSERT_FALSE(commit_done_at_recovery_end);
+  ASSERT_TRUE(synced);
+  ASSERT_OK(commit_status);
+  std::string value;
+  ASSERT_OK(db.Get(ReadOptions(), "commit_key", &value));
+  ASSERT_EQ("commit_value", value);
+
+  wal_only_commit_txn.reset();
+  Close();
+}
+
 // Regression test for the gap left by syncing only at the start of
 // ResumeImpl(). Recovery drops mutex_ in FlushAllColumnFamilies() before it
 // enters AtomicFlushMemTables(), so a WriteUnprepared WAL-only commit can