Add remote compaction format compatibility test to check_format_compatible.sh

Rename db_test2 -> db_etc2_test (#14218 )
Summary: When searching/grepping through unit tests, it's convenient to use test.cc suffix to match all unit tests, or to exclude test.cc when excluding unit tests from a code search. (I've even seen my AI assistant grep through `*test.cc`.) So I'm renaming db_test2.cc (as planned in https://github.com/facebook/rocksdb/issues/14076) Pull Request resolved: https://github.com/facebook/rocksdb/pull/14218 Test Plan: existing tests + CI Reviewed By: xingbowang Differential Revision: D90140624 Pulled By: pdillinger fbshipit-source-id: 78aa099290670bbb4093b2c7c02bc47ab3bf5f8e
2026-07-07 14:47:40 +08:00 · 2026-06-03 08:48:11 -07:00 · 2026-06-02 16:18:05 -07:00 · 2026-06-02 11:13:54 -07:00 · 2026-06-01 18:13:29 -07:00 · 2026-06-01 17:07:11 -07:00
42 changed files with 1347 additions and 157 deletions
@@ -3,7 +3,7 @@ inputs:
  suite-run:
    description: Comma-separated list of test suites to run (empty to skip C++ tests)
    required: false
-    default: arena_test,db_basic_test,db_test,db_test2,db_merge_operand_test,bloom_test,c_test,coding_test,crc32c_test,dynamic_bloom_test,env_basic_test,env_test,hash_test,random_test
+    default: arena_test,db_basic_test,db_test,db_etc2_test,db_merge_operand_test,bloom_test,c_test,coding_test,crc32c_test,dynamic_bloom_test,env_basic_test,env_test,hash_test,random_test
  run-java:
    description: Whether to run Java tests
    required: false
@@ -537,7 +537,7 @@ jobs:
          suite_run: db_test
          run_java: "false"
        - test_shard: other
-          suite_run: arena_test,db_basic_test,db_test2,db_merge_operand_test,bloom_test,c_test,coding_test,crc32c_test,dynamic_bloom_test,env_basic_test,env_test,hash_test,random_test
+          suite_run: arena_test,db_basic_test,db_etc2_test,db_merge_operand_test,bloom_test,c_test,coding_test,crc32c_test,dynamic_bloom_test,env_basic_test,env_test,hash_test,random_test
          run_java: "false"
        - test_shard: java
          suite_run: ""
@@ -52,7 +52,7 @@ GRTAGS
 GTAGS
 rocksdb_dump
 rocksdb_undump
-db_test2
+db_etc2_test
 trace_analyzer
 block_cache_trace_analyzer
 io_tracer_parser
@@ -4868,6 +4868,12 @@ cpp_unittest_wrapper(name="db_encryption_test",
            extra_compiler_flags=[])


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


-cpp_unittest_wrapper(name="db_test2",
-            srcs=["db/db_test2.cc"],
-            deps=[":rocksdb_test_lib"],
-            extra_compiler_flags=[])
-
-
 cpp_unittest_wrapper(name="db_universal_compaction_test",
            srcs=["db/db_universal_compaction_test.cc"],
            deps=[":rocksdb_test_lib"],
@@ -209,12 +209,11 @@ The following patterns emerged as frequent sources of review feedback:
 ## Important tips

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

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

-db_test2: $(OBJ_DIR)/db/db_test2.o $(TEST_LIBRARY) $(LIBRARY)
+db_etc2_test: $(OBJ_DIR)/db/db_etc2_test.o $(TEST_LIBRARY) $(LIBRARY)
 	$(AM_LINK)

 db_etc3_test: $(OBJ_DIR)/db/db_etc3_test.o $(TEST_LIBRARY) $(LIBRARY)
@@ -651,6 +651,10 @@ static std::unordered_map<std::string, OptionTypeInfo>
        {"cpu_micros",
         {offsetof(struct CompactionJobStats, cpu_micros), OptionType::kUInt64T,
          OptionVerificationType::kNormal, OptionTypeFlags::kNone}},
+        {"has_accurate_num_input_records",
+         {offsetof(struct CompactionJobStats, has_accurate_num_input_records),
+          OptionType::kBoolean, OptionVerificationType::kNormal,
+          OptionTypeFlags::kNone}},
        {"num_input_records",
         {offsetof(struct CompactionJobStats, num_input_records),
          OptionType::kUInt64T, OptionVerificationType::kNormal,
@@ -943,6 +943,37 @@ TEST_F(CompactionServiceTest, VerifyInputRecordCount) {
  SyncPoint::GetInstance()->ClearAllCallBacks();
 }

+TEST_F(CompactionServiceTest, InaccurateInputRecordCount) {
+  Options options = CurrentOptions();
+  options.disable_auto_compactions = true;
+  ReopenWithCompactionService(&options);
+  GenerateTestData();
+
+  auto my_cs = GetCompactionService();
+
+  std::string start_str = Key(15);
+  std::string end_str = Key(45);
+  Slice start(start_str);
+  Slice end(end_str);
+  uint64_t comp_num = my_cs->GetCompactionNum();
+
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack(
+      "CompactionServiceCompactionJob::Run:0", [&](void* arg) {
+        CompactionServiceResult* compaction_result =
+            *(static_cast<CompactionServiceResult**>(arg));
+        ASSERT_TRUE(compaction_result != nullptr);
+        compaction_result->stats.has_accurate_num_input_records = false;
+        compaction_result->stats.num_input_records = 0;
+      });
+  SyncPoint::GetInstance()->EnableProcessing();
+
+  ASSERT_OK(db_->CompactRange(CompactRangeOptions(), &start, &end));
+  ASSERT_GE(my_cs->GetCompactionNum(), comp_num + 1);
+
+  SyncPoint::GetInstance()->DisableProcessing();
+  SyncPoint::GetInstance()->ClearAllCallBacks();
+}
+
 TEST_F(CompactionServiceTest, EmptyResult) {
  Options options = CurrentOptions();
  options.disable_auto_compactions = true;
@@ -2022,7 +2022,11 @@ TEST_F(DBBloomFilterTest, MutableFilterPolicy) {
  double expected_bpk = 10.0;
  // Other configs to try
  std::vector<std::pair<std::string, double>> configs = {
-      {"ribbonfilter:10:-1", 7.0}, {"bloomfilter:5", 5.0}, {"nullptr", 0.0}};
+      {"ribbonfilter:10:-1", 7.0},
+      {"bloomfilter:5", 5.0},
+      {"nullptr", 0.0},
+      // As serialized in OPTIONS file
+      {"{id=ribbonfilter:12:-1;bloom_before_level=-1;}", 8.4}};

  table_options.cache_index_and_filter_blocks = true;
  options.table_factory.reset(NewBlockBasedTableFactory(table_options));
@@ -427,6 +427,38 @@ TEST_F(DBCompactionAbortTest, AbortAutomaticCompaction) {
  }
 }

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

 class DBTest2 : public DBTestBase {
 public:
-  DBTest2() : DBTestBase("db_test2", /*env_do_fsync=*/true) {}
+  DBTest2() : DBTestBase("db_etc2_test", /*env_do_fsync=*/true) {}
 };

 TEST_F(DBTest2, OpenForReadOnly) {
@@ -4,6 +4,9 @@
 //  (found in the LICENSE.Apache file in the root directory).

 #include "db/db_test_util.h"
+#include "rocksdb/convenience.h"
+#include "rocksdb/metadata.h"
+#include "rocksdb/sst_file_writer.h"

 namespace ROCKSDB_NAMESPACE {

@@ -43,50 +46,98 @@ TEST_F(DBEtc3Test, AutoTuneManifestSize) {
  ASSERT_EQ(DBOptions{}.max_manifest_space_amp_pct, 500);

  Options options = CurrentOptions();
-  ASSERT_OK(db_->SetOptions({{"level0_file_num_compaction_trigger", "20"}}));
+  // Test setup: create many flushed files. Keep level compaction semantics so
+  // DeleteFilesInRanges can remove non-L0 files, but prevent automatic
+  // compactions and write stalls from adding unrelated behavior.
+  options.disable_auto_compactions = true;
+  options.level0_slowdown_writes_trigger = 100;
+  options.level0_stop_writes_trigger = 200;

-  // Use large column family names to essentially control the amount of payload
-  // data needed for the manifest file. Drop manifest entries don't include the
-  // CF name so are small.
+  // Test strategy: use large column family names to control the rough amount
+  // of payload added to the MANIFEST. Drop manifest entries do not include the
+  // CF name, so they are small.
+  //
+  // Most CF helper calls piggy-back a background manifest write so the main
+  // auto-tuning phases continue to test the unrelaxed background threshold
+  // even though CF manipulation itself is foreground. Phase-specific foreground
+  // checks disable that piggy-backed background write.
  uint64_t prev_manifest_num = 0, cur_manifest_num = 0;
  std::deque<ColumnFamilyHandle*> handles;
  int counter = 5;
-  auto AddCfFn = [&]() {
+  auto UpdateManifestNumsFrom = [&](uint64_t before_manifest_num) {
+    prev_manifest_num = before_manifest_num;
+    cur_manifest_num = dbfull()->TEST_Current_Manifest_FileNo();
+  };
+  auto BackgroundManifestWriteFn = [&]() {
+    uint64_t before_manifest_num = cur_manifest_num;
+    ASSERT_OK(Put("x", std::to_string(counter++)));
+    ASSERT_OK(Flush());
+    UpdateManifestNumsFrom(before_manifest_num);
+  };
+  auto AddCfFn = [&](bool include_background_manifest_write = true) {
+    uint64_t before_manifest_num = cur_manifest_num;
    std::string name = "cf" + std::to_string(counter++);
    name.resize(1000, 'a');
    ASSERT_OK(db_->CreateColumnFamily(options, name, &handles.emplace_back()));
-    prev_manifest_num = cur_manifest_num;
    cur_manifest_num = dbfull()->TEST_Current_Manifest_FileNo();
+    if (include_background_manifest_write) {
+      BackgroundManifestWriteFn();
+    }
+    UpdateManifestNumsFrom(before_manifest_num);
  };
-  auto DropCfFn = [&]() {
+  auto DropCfFn = [&](bool include_background_manifest_write = true) {
+    uint64_t before_manifest_num = cur_manifest_num;
    ASSERT_OK(db_->DropColumnFamily(handles.front()));
    ASSERT_OK(db_->DestroyColumnFamilyHandle(handles.front()));
    handles.pop_front();
-    prev_manifest_num = cur_manifest_num;
-    cur_manifest_num = dbfull()->TEST_Current_Manifest_FileNo();
-  };
-  auto TrivialManifestWriteFn = [&]() {
-    ASSERT_OK(Put("x", std::to_string(counter++)));
-    ASSERT_OK(Flush());
-    prev_manifest_num = cur_manifest_num;
    cur_manifest_num = dbfull()->TEST_Current_Manifest_FileNo();
+    if (include_background_manifest_write) {
+      BackgroundManifestWriteFn();
+    }
+    UpdateManifestNumsFrom(before_manifest_num);
  };

+  // ---- Phase 1: foreground threshold relaxation is bounded ----
+  //
+  // Foreground operations should only get about 25% extra headroom, not an
+  // unbounded threshold. With 1000-byte CF names and a 3000-byte normal limit,
+  // the relaxed limit should allow the first four foreground-only CF additions
+  // but require rotation on the fifth.
+  options.max_manifest_file_size = 3000;
+  options.max_manifest_space_amp_pct = 0;
+  DestroyAndReopen(options);
+  cur_manifest_num = dbfull()->TEST_Current_Manifest_FileNo();
+  prev_manifest_num = cur_manifest_num;
+  for (int i = 1; i <= 4; ++i) {
+    AddCfFn(/*include_background_manifest_write=*/false);
+    ASSERT_EQ(prev_manifest_num, cur_manifest_num);
+  }
+  AddCfFn(/*include_background_manifest_write=*/false);
+  ASSERT_LT(prev_manifest_num, cur_manifest_num);
+  while (!handles.empty()) {
+    ASSERT_OK(db_->DestroyColumnFamilyHandle(handles.front()));
+    handles.pop_front();
+  }
+
+  // ---- Phase 2: no auto-tuning means frequent rotation ----
+  //
  options.max_manifest_file_size = 1000000;
  options.max_manifest_space_amp_pct = 0;  // no auto-tuning yet
  DestroyAndReopen(options);
+  cur_manifest_num = dbfull()->TEST_Current_Manifest_FileNo();
+  prev_manifest_num = cur_manifest_num;

-  // With the generous (minimum) maximum manifest size, should not be rotated
+  // With the generous minimum manifest size, should not be rotated.
  AddCfFn();
  AddCfFn();
  AddCfFn();
  ASSERT_EQ(prev_manifest_num, cur_manifest_num);

-  // Change options for small max and (still) no auto-tuning
+  // Lower the minimum while still disabling auto-tuning.
  ASSERT_OK(db_->SetDBOptions({{"max_manifest_file_size", "3000"}}));

  // Takes effect on the next manifest write
-  TrivialManifestWriteFn();
+  BackgroundManifestWriteFn();
  ASSERT_LT(prev_manifest_num, cur_manifest_num);

  // Now we have to rewrite the whole manifest on each write because the
@@ -97,28 +148,31 @@ TEST_F(DBEtc3Test, AutoTuneManifestSize) {
  ASSERT_LT(prev_manifest_num, cur_manifest_num);
  AddCfFn();
  ASSERT_LT(prev_manifest_num, cur_manifest_num);
-  TrivialManifestWriteFn();
+  BackgroundManifestWriteFn();
  ASSERT_LT(prev_manifest_num, cur_manifest_num);

+  // ---- Phase 3: auto-tuning raises the background threshold ----
+  //
  // Enabling auto-tuning should fix this, immediately for next manifest writes.
-  // This will allow up to double-ish the size of the compacted manifest,
-  // which last should have been 4000 + some bytes.
+  // This will allow up to roughly double the size of the compacted manifest,
+  // which now includes CF entries plus the piggy-backed background writes.
  ASSERT_EQ(handles.size(), 4U);
-  ASSERT_OK(db_->SetDBOptions({{"max_manifest_space_amp_pct", "105"}}));
+  ASSERT_OK(db_->SetDBOptions({{"max_manifest_space_amp_pct", "95"}}));

-  // After 9 CF names should be enough to rotate the manifest
-  for (int i = 1; i <= 5; ++i) {
+  // Auto-tuning lets several more CF names accumulate in the MANIFEST before
+  // the piggy-backed background write crosses the threshold.
+  for (int i = 1; i <= 3; ++i) {
    if ((i % 2) == 1) {
      DropCfFn();
    }
    AddCfFn();
    ASSERT_EQ(prev_manifest_num, cur_manifest_num);
  }
-  TrivialManifestWriteFn();
+  AddCfFn();
  ASSERT_LT(prev_manifest_num, cur_manifest_num);

-  // We now have a different last compacted manifest size, should be
-  // able to go beyond 9 CFs named in manifest this time.
+  // We now have a different last compacted manifest size, so the next
+  // threshold should be based on the newly compacted MANIFEST.
  ASSERT_EQ(handles.size(), 6U);

  DropCfFn();
@@ -128,11 +182,10 @@ TEST_F(DBEtc3Test, AutoTuneManifestSize) {
    AddCfFn();
    ASSERT_EQ(prev_manifest_num, cur_manifest_num);
  }
-  // We've written 10 named CFs to the manifest. We should be able to
-  // dynamically change the auto-tuning still based on the last "compacted"
-  // manifest size of 7000 + some bytes.
+  // We should be able to dynamically change the auto-tuning still based on
+  // the last "compacted" manifest size.
  ASSERT_OK(db_->SetDBOptions({{"max_manifest_space_amp_pct", "51"}}));
-  TrivialManifestWriteFn();
+  BackgroundManifestWriteFn();
  ASSERT_LT(prev_manifest_num, cur_manifest_num);
  // And the "compacted" manifest size has reset again, so should be changed
  // again sooner.
@@ -141,16 +194,129 @@ TEST_F(DBEtc3Test, AutoTuneManifestSize) {
    AddCfFn();
    ASSERT_EQ(prev_manifest_num, cur_manifest_num);
  }
-  // Enough for manifest change
-  AddCfFn();
+
+  // ---- Phase 4: foreground operations use relaxed threshold ----
+  //
+  // The current MANIFEST is now large enough for the next background manifest
+  // write to rotate it, but still small enough for foreground operations to use
+  // their 25% extra headroom. Assert that each foreground operation stays on
+  // the same MANIFEST, then verify the next background write rotates.
+  const std::string external_files_dir = dbname_ + "/external_files";
+  ASSERT_OK(env_->CreateDirIfMissing(external_files_dir));
+  auto WriteExternalSstFile = [&](const std::string& file_name,
+                                  const std::string& key,
+                                  const std::string& value) {
+    const std::string file_path = external_files_dir + "/" + file_name;
+    Status ignored = env_->DeleteFile(file_path);
+    ignored.PermitUncheckedError();
+
+    SstFileWriter sst_file_writer(EnvOptions(), options);
+    ASSERT_OK(sst_file_writer.Open(file_path));
+    ASSERT_OK(sst_file_writer.Put(key, value));
+    ASSERT_OK(sst_file_writer.Finish());
+  };
+  auto IngestExternalFileForegroundManifestWriteFn =
+      [&](std::string* ingested_key) {
+        uint64_t before_manifest_num = cur_manifest_num;
+        const std::string key = "z" + std::to_string(counter++);
+        const std::string value = "v" + std::to_string(counter++);
+        const std::string file_name =
+            "ingest_" + std::to_string(counter++) + ".sst";
+        const std::string file_path = external_files_dir + "/" + file_name;
+        WriteExternalSstFile(file_name, key, value);
+
+        ASSERT_OK(
+            db_->IngestExternalFile({file_path}, IngestExternalFileOptions()));
+        ASSERT_EQ(value, Get(key));
+        *ingested_key = key;
+        Status ignored = env_->DeleteFile(file_path);
+        ignored.PermitUncheckedError();
+        UpdateManifestNumsFrom(before_manifest_num);
+      };
+  auto CreateColumnFamilyWithImportForegroundManifestWriteFn = [&]() {
+    uint64_t before_manifest_num = cur_manifest_num;
+    const std::string cf_name = "import_cf" + std::to_string(counter++);
+    const std::string key = "import" + std::to_string(counter++);
+    const std::string value = "v" + std::to_string(counter++);
+    const std::string file_name =
+        "import_" + std::to_string(counter++) + ".sst";
+    const std::string file_path = external_files_dir + "/" + file_name;
+    WriteExternalSstFile(file_name, key, value);
+
+    LiveFileMetaData file_metadata;
+    file_metadata.name = file_name;
+    file_metadata.db_path = external_files_dir;
+    file_metadata.smallest_seqno = 0;
+    file_metadata.largest_seqno = 0;
+    file_metadata.level = 0;
+    ExportImportFilesMetaData metadata;
+    metadata.files.push_back(file_metadata);
+    metadata.db_comparator_name = options.comparator->Name();
+
+    ColumnFamilyHandle* import_handle = nullptr;
+    ASSERT_OK(db_->CreateColumnFamilyWithImport(options, cf_name,
+                                                ImportColumnFamilyOptions(),
+                                                metadata, &import_handle));
+    ASSERT_NE(import_handle, nullptr);
+    handles.push_back(import_handle);
+
+    std::string result;
+    ASSERT_OK(db_->Get(ReadOptions(), import_handle, key, &result));
+    ASSERT_EQ(value, result);
+    Status ignored = env_->DeleteFile(file_path);
+    ignored.PermitUncheckedError();
+    UpdateManifestNumsFrom(before_manifest_num);
+  };
+  auto DeleteFilesInRangesForegroundManifestWriteFn =
+      [&](const std::string& key) {
+        uint64_t before_manifest_num = cur_manifest_num;
+        const std::string limit = key + "\xff";
+        std::vector<RangeOpt> ranges;
+        ranges.emplace_back(key, limit);
+        ASSERT_OK(DeleteFilesInRanges(db_.get(), db_->DefaultColumnFamily(),
+                                      ranges.data(), ranges.size(),
+                                      /*include_end=*/false));
+
+        std::string result;
+        ASSERT_TRUE(db_->Get(ReadOptions(), key, &result).IsNotFound());
+        UpdateManifestNumsFrom(before_manifest_num);
+      };
+
+  // Column family manipulation.
+  AddCfFn(/*include_background_manifest_write=*/false);
+  ASSERT_EQ(prev_manifest_num, cur_manifest_num);
+
+  // SetOptions should not write to the MANIFEST. If that regresses, the write
+  // should be treated as a background write and rotate here.
+  {
+    ASSERT_OK(db_->SetOptions({{"level0_slowdown_writes_trigger", "101"}}));
+    UpdateManifestNumsFrom(cur_manifest_num);
+  }
+  ASSERT_EQ(prev_manifest_num, cur_manifest_num);
+
+  // External file ingestion.
+  std::string ingested_key;
+  IngestExternalFileForegroundManifestWriteFn(&ingested_key);
+  ASSERT_EQ(prev_manifest_num, cur_manifest_num);
+
+  // Imported column family creation.
+  CreateColumnFamilyWithImportForegroundManifestWriteFn();
+  ASSERT_EQ(prev_manifest_num, cur_manifest_num);
+
+  // Physical file deletion by range.
+  DeleteFilesInRangesForegroundManifestWriteFn(ingested_key);
+  ASSERT_EQ(prev_manifest_num, cur_manifest_num);
+
+  // Background flush.
+  BackgroundManifestWriteFn();
  ASSERT_LT(prev_manifest_num, cur_manifest_num);

-  // ---- Verify persisted compacted manifest size survives close/reopen ----
+  // ---- Phase 5: persisted compacted manifest size survives close/reopen ----
  // Close with CFs still live. Reopen with reuse_manifest_on_open so the
  // manifest is NOT rewritten from scratch. The persisted compacted size
  // should be loaded and used for auto-tuning.
-  // At this point we have 7 CF handles plus default.
-  ASSERT_EQ(handles.size(), 7U);
+  // At this point we have 8 CF handles plus default.
+  ASSERT_EQ(handles.size(), 8U);

  // Collect CF names for reopen, then release handles (Close needs this)
  std::vector<std::string> cf_names = {"default"};
@@ -163,18 +329,16 @@ TEST_F(DBEtc3Test, AutoTuneManifestSize) {
  handles.clear();

  Close();
-  // Use a large max_manifest_file_size so the reused manifest (which is
-  // already ~10KB) does NOT trigger rotation on the first few writes.
-  // Auto-tuning with the persisted compacted size (~5KB) at 200% amp
-  // gives a tuned threshold of ~15KB. Without persistence, the threshold
-  // would be max(max_manifest_file_size, 0 * anything) =
-  // max_manifest_file_size.
+  // Use a max_manifest_file_size below the reused manifest size. Auto-tuning
+  // with the persisted compacted size at 200% amp keeps the tuned threshold
+  // high enough. Without persistence, the threshold would be
+  // max(max_manifest_file_size, 0 * anything) = max_manifest_file_size.
  //
-  // We set max_manifest_file_size to two values to distinguish:
-  // - 3000: if persisted compacted size is NOT loaded, tuned = 3000,
-  //   and the first AddCf will rotate (manifest is already ~10KB > 3000)
-  // - With persisted compacted size loaded, tuned = max(3000, 5000*3) = 15000,
-  //   so no rotation until we exceed 15KB
+  // With max_manifest_file_size set to 3000, missing persisted compacted size
+  // would keep tuned = 3000 and the first AddCf would rotate because the
+  // reused manifest is already larger. With persisted compacted size loaded,
+  // the tuned threshold is based on the compacted size, so no rotation until
+  // the manifest grows well beyond the minimum.
  options.max_manifest_file_size = 3000;
  options.max_manifest_space_amp_pct = 200;
  options.reuse_manifest_on_open = true;
@@ -186,11 +350,11 @@ TEST_F(DBEtc3Test, AutoTuneManifestSize) {
  // during reopen, because the persisted compacted size keeps the tuned
  // threshold high enough. Without persistence, last_compacted = 0, so
  // tuned = max_manifest_file_size = 3000, and the first LogAndApply
-  // during Open rotates the manifest because it's already ~10KB > 3000.
+  // during Open rotates the manifest because it is already larger than 3000.
  ASSERT_EQ(manifest_num_before_reopen, cur_manifest_num);

-  // Adding CFs should still not trigger rotation because the tuned
-  // threshold (~15KB) exceeds the current manifest size (~10KB + adds).
+  // Adding CFs should still not trigger rotation because the tuned threshold
+  // from the persisted compacted size exceeds the current manifest size.
  for (int i = 1; i <= 4; ++i) {
    AddCfFn();
    ASSERT_EQ(prev_manifest_num, cur_manifest_num);
@@ -202,7 +366,7 @@ TEST_F(DBEtc3Test, AutoTuneManifestSize) {

  // Wrap up
  while (!handles.empty()) {
-    DropCfFn();
+    DropCfFn(/*include_background_manifest_write=*/false);
  }
 }

@@ -5853,6 +5853,7 @@ Status DBImpl::DeleteFilesInRanges(ColumnFamilyHandle* column_family,
  }

  VersionEdit edit;
+  edit.MarkForegroundOperation();
  std::set<FileMetaData*> deleted_files;
  JobContext job_context(next_job_id_.fetch_add(1), true);
  {
@@ -6963,7 +6964,9 @@ Status DBImpl::IngestExternalFiles(
        assert(!cfd->IsDropped());
        cfds_to_commit.push_back(cfd);
        autovector<VersionEdit*> edit_list;
-        edit_list.push_back(ingestion_jobs[i].edit());
+        auto* edit = ingestion_jobs[i].edit();
+        edit->MarkForegroundOperation();
+        edit_list.push_back(edit);
        edit_lists.push_back(edit_list);
        ++num_entries;
      }
@@ -6978,7 +6981,6 @@ Status DBImpl::IngestExternalFiles(
      }
      status =
          versions_->LogAndApply(cfds_to_commit, read_options, write_options,
-
                                 edit_lists, &mutex_, directories_.GetDbDir());
      // It is safe to update VersionSet last seqno here after LogAndApply since
      // LogAndApply persists last sequence number from VersionEdits,
@@ -7111,6 +7113,7 @@ Status DBImpl::CreateColumnFamilyWithImport(

  SuperVersionContext dummy_sv_ctx(/* create_superversion */ true);
  VersionEdit dummy_edit;
+  dummy_edit.MarkForegroundOperation();
  uint64_t next_file_number = 0;
  std::unique_ptr<std::list<uint64_t>::iterator> pending_output_elem;
  {
@@ -7168,9 +7171,10 @@ Status DBImpl::CreateColumnFamilyWithImport(

      // Install job edit [Mutex will be unlocked here]
      if (status.ok()) {
-        status = versions_->LogAndApply(cfd, read_options, write_options,
-                                        import_job.edit(), &mutex_,
-                                        directories_.GetDbDir());
+        auto* edit = import_job.edit();
+        edit->MarkForegroundOperation();
+        status = versions_->LogAndApply(cfd, read_options, write_options, edit,
+                                        &mutex_, directories_.GetDbDir());
        if (status.ok()) {
          InstallSuperVersionForConfigChange(cfd, &sv_context);
        }
@@ -7601,6 +7605,7 @@ Status DBImpl::ReserveFileNumbersBeforeIngestion(
  // reuse the file number that has already assigned to the internal file,
  // and this will overwrite the external file. To protect the external
  // file, we have to make sure the file number will never being reused.
+  dummy_edit.MarkForegroundOperation();
  s = versions_->LogAndApply(cfd, read_options, write_options, &dummy_edit,
                             &mutex_, directories_.GetDbDir());
  if (s.ok()) {
@@ -4067,6 +4067,13 @@ Status DBImpl::BackgroundCompaction(bool* made_progress,
      status = Status::ShutdownInProgress();
    } else if (compaction_aborted_.load(std::memory_order_acquire) > 0) {
      status = Status::Incomplete(Status::SubCode::kCompactionAborted);
+      if (!is_prepicked) {
+        // This automatic compaction was scheduled from compaction_queue_, but
+        // abort happened before PickCompactionFromQueue() could remove the
+        // queued CF. Restore the unscheduled count so ResumeAllCompactions()
+        // can schedule the still-queued work.
+        unscheduled_compactions_++;
+      }
    } else if (is_manual &&
               manual_compaction->canceled.load(std::memory_order_acquire)) {
      status = Status::Incomplete(Status::SubCode::kManualCompactionPaused);
@@ -4074,10 +4081,14 @@ Status DBImpl::BackgroundCompaction(bool* made_progress,
  } else {
    status = error_handler_.GetBGError();
    // If we get here, it means a hard error happened after this compaction
-    // was scheduled by MaybeScheduleFlushOrCompaction(), but before it got
-    // a chance to execute. Since we didn't pop a cfd from the compaction
-    // queue, increment unscheduled_compactions_
-    unscheduled_compactions_++;
+    // was scheduled by MaybeScheduleFlushOrCompaction(), but before it got a
+    // chance to execute. Since non-prepicked work consumed an unscheduled
+    // compaction credit without popping a cfd from the compaction queue,
+    // restore the credit here. Prepicked work was scheduled directly and did
+    // not consume unscheduled_compactions_.
+    if (!is_prepicked) {
+      unscheduled_compactions_++;
+    }
  }

  if (!status.ok()) {
@@ -216,12 +216,19 @@ class TestIterator : public InternalIterator {
  bool IsKeyPinned() const override { return true; }
  bool IsValuePinned() const override { return true; }

+  void Prepare(const MultiScanArgs* /*scan_opts*/) override {
+    ++prepare_call_count_;
+  }
+
+  size_t prepare_call_count() const { return prepare_call_count_; }
+
 private:
  bool initialized_;
  bool valid_;
  size_t sequence_number_;
  size_t iter_;
  size_t steps_ = 0;
+  size_t prepare_call_count_ = 0;

  InternalKeyComparator cmp;
  std::vector<std::pair<std::string, std::string>> data_;
@@ -243,6 +250,53 @@ class DBIteratorTest : public testing::Test {
  DBIteratorTest() : env_(Env::Default()) {}
 };

+TEST_F(DBIteratorTest, PrepareForwardsValidatedScanRanges) {
+  Options options;
+  ImmutableOptions ioptions = ImmutableOptions(options);
+  MutableCFOptions mutable_cf_options = MutableCFOptions(options);
+  TestIterator* internal_iter = new TestIterator(BytewiseComparator());
+  internal_iter->AddPut("a", "val_a");
+  internal_iter->AddPut("b", "val_b");
+  internal_iter->Finish();
+
+  ReadOptions ro;
+  std::unique_ptr<Iterator> db_iter(DBIter::NewIter(
+      env_, ro, ioptions, mutable_cf_options, BytewiseComparator(),
+      internal_iter, nullptr /* version */, 10 /* sequence */,
+      nullptr /* read_callback */, /*active_mem=*/nullptr));
+
+  MultiScanArgs scan_opts(BytewiseComparator());
+  scan_opts.insert("a", "c");
+
+  db_iter->Prepare(scan_opts);
+  ASSERT_OK(db_iter->status());
+  EXPECT_EQ(internal_iter->prepare_call_count(), 1);
+}
+
+TEST_F(DBIteratorTest, PrepareDoesNotForwardInvalidScanRanges) {
+  Options options;
+  ImmutableOptions ioptions = ImmutableOptions(options);
+  MutableCFOptions mutable_cf_options = MutableCFOptions(options);
+  TestIterator* internal_iter = new TestIterator(BytewiseComparator());
+  internal_iter->AddPut("a", "val_a");
+  internal_iter->AddPut("b", "val_b");
+  internal_iter->Finish();
+
+  ReadOptions ro;
+  std::unique_ptr<Iterator> db_iter(DBIter::NewIter(
+      env_, ro, ioptions, mutable_cf_options, BytewiseComparator(),
+      internal_iter, nullptr /* version */, 10 /* sequence */,
+      nullptr /* read_callback */, /*active_mem=*/nullptr));
+
+  MultiScanArgs scan_opts(BytewiseComparator());
+  scan_opts.insert("b", "d");
+  scan_opts.insert("a", "c");
+
+  db_iter->Prepare(scan_opts);
+  EXPECT_TRUE(db_iter->status().IsInvalidArgument());
+  EXPECT_EQ(internal_iter->prepare_call_count(), 0);
+}
+
 TEST_F(DBIteratorTest, DBIteratorPrevNext) {
  Options options;
  ImmutableOptions ioptions = ImmutableOptions(options);
@@ -4395,6 +4395,71 @@ TEST_P(DBMultiScanIteratorTest, RangeAcrossFiles) {
  iter.reset();
 }

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

  bool IsColumnFamilyDrop() const { return is_column_family_drop_; }

+  void MarkForegroundOperation() { is_foreground_operation_ = true; }
+  bool IsForegroundOperation() const {
+    return is_foreground_operation_ || IsColumnFamilyManipulation();
+  }
+
  void MarkNoManifestWriteDummy() { is_no_manifest_write_dummy_ = true; }
  bool IsNoManifestWriteDummy() const { return is_no_manifest_write_dummy_; }

@@ -1111,6 +1116,7 @@ class VersionEdit {
  // it also includes column family name.
  bool is_column_family_drop_ = false;
  bool is_column_family_add_ = false;
+  bool is_foreground_operation_ = false;
  std::string column_family_name_;

  uint32_t remaining_entries_ = 0;
@@ -6111,9 +6111,32 @@ Status VersionSet::ProcessManifestWrites(

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

+// StringToMap parses a serialized options string into a map. Each
+// resulting map value is in a self-contained form (it can be embedded
+// directly in a `key=value;` context -- e.g. SetOptions -- without further
+// escaping). Specifically: nested braced values from the input are
+// preserved with their outer braces. Permissive: accepts both braced and
+// unbraced forms; values not requiring braces are returned as-is.
+// Example:
+//   "filter_policy={id=ribbonfilter:10;bloom_before_level=-1};block_size=4096"
+// produces:
+//   {filter_policy -> "{id=ribbonfilter:10;bloom_before_level=-1}",
+//    block_size    -> "4096"}
 Status StringToMap(const std::string& opts_str,
                   std::unordered_map<std::string, std::string>* opts_map);

+// MapToString is the inverse of StringToMap: a naive `key=value;` join.
+// Each map value must already be in self-contained form (as returned by
+// StringToMap) -- i.e. simple text or a single balanced `{...}` block.
+// Values from StringToMap satisfy this property, so the round-trip
+//
+//   StringToMap(MapToString(StringToMap(s))) == StringToMap(s)
+//
+// holds. Callers building a map by hand are responsible for ensuring
+// values are self-contained; raw values containing `;` or starting with
+// `{` without matching `}` won't round-trip.
+Status MapToString(const std::unordered_map<std::string, std::string>& opts_map,
+                   std::string* opts_str);
+
 // Request stopping background work, if wait is true wait until it's done
 void CancelAllBackgroundWork(DB* db, bool wait = false);

@@ -152,6 +152,11 @@ class BlockBasedTable;
 struct JobOptions {
  uint64_t io_coalesce_threshold = 16 * 1024;
  ReadOptions read_options;
+
+  // True when IOJob::block_handles is already sorted by file offset. Callers
+  // that can guarantee sorted order may set this to let IODispatcher skip
+  // defensive sorting before cache lookup and coalescing.
+  bool block_handles_are_sorted = false;
 };

 class IOJob {
@@ -1003,6 +1003,13 @@ struct DBOptions {
  //
  // This option is mutable with SetDBOptions(), taking effect on the next
  // manifest write (e.g. completed DB compaction or flush).
+  //
+  // For MANIFEST write batches containing foreground operations like external
+  // file ingestion/import, DeleteFilesInRange, CreateColumnFamily, and
+  // DropColumnFamily, the effective limit is relaxed by 25% to reduce the
+  // likelihood of user operations blocking on MANIFEST rotation.
+  // Background-only batches (flush and compaction) use the configured or
+  // auto-tuned limit directly.
  uint64_t max_manifest_file_size = 1024 * 1024 * 1024;

  // If true, on DB close, read back the entire MANIFEST file and validate
@@ -453,11 +453,13 @@ class OptionTypeInfo {
                                       const std::string& value, void* addr) {
          std::map<std::string, std::string> map;
          Status s;
+          // Permit `{k1=v1;k2=v2}` (as serialized) or bare `k1=v1;k2=v2`.
+          std::string stripped = OptionTypeInfo::StripOuterBraces(value);
          for (size_t start = 0, end = 0;
-               s.ok() && start < value.size() && end != std::string::npos;
+               s.ok() && start < stripped.size() && end != std::string::npos;
               start = end + 1) {
            std::string token;
-            s = OptionTypeInfo::NextToken(value, item_separator, start, &end,
+            s = OptionTypeInfo::NextToken(stripped, item_separator, start, &end,
                                          &token);
            if (s.ok() && !token.empty()) {
              size_t pos = token.find(kv_separator);
@@ -947,6 +949,14 @@ class OptionTypeInfo {
  static Status NextToken(const std::string& opts, char delimiter, size_t start,
                          size_t* end, std::string* token);

+  // Strips a single layer of outer "{" / "}" wrapping iff the leading '{'
+  // pairs with the trailing '}'. This permissively unwraps values that
+  // could legitimately be either braced or bare (e.g. `{a:b:c}` and
+  // `a:b:c` are accepted equivalently by list-style parsers). The check
+  // is done at brace depth: `{a}:{b}` is left as-is because the leading
+  // '{' closes before the end of the string.
+  static std::string StripOuterBraces(const std::string& value);
+
  constexpr static const char* kIdPropName() { return "id"; }
  constexpr static const char* kIdPropSuffix() { return ".id"; }

@@ -995,12 +1005,16 @@ Status ParseArray(const ConfigOptions& config_options,

  ConfigOptions copy = config_options;
  copy.ignore_unsupported_options = false;
+  // Permit the caller to pass either bare `item1:item2` or wrapped
+  // `{item1:item2}`. Either form parses identically.
+  std::string stripped = OptionTypeInfo::StripOuterBraces(value);
  size_t i = 0, start = 0, end = 0;
-  for (; status.ok() && i < kSize && start < value.size() &&
+  for (; status.ok() && i < kSize && start < stripped.size() &&
         end != std::string::npos;
       i++, start = end + 1) {
    std::string token;
-    status = OptionTypeInfo::NextToken(value, separator, start, &end, &token);
+    status =
+        OptionTypeInfo::NextToken(stripped, separator, start, &end, &token);
    if (status.ok()) {
      status = elem_info.Parse(copy, name, token, &((*result)[i]));
      if (config_options.ignore_unsupported_options &&
@@ -1137,11 +1151,15 @@ Status ParseVector(const ConfigOptions& config_options,
  // object is valid or not.
  ConfigOptions copy = config_options;
  copy.ignore_unsupported_options = false;
+  // Permit the caller to pass either bare `item1:item2` or wrapped
+  // `{item1:item2}`. Either form parses identically.
+  std::string stripped = OptionTypeInfo::StripOuterBraces(value);
  for (size_t start = 0, end = 0;
-       status.ok() && start < value.size() && end != std::string::npos;
+       status.ok() && start < stripped.size() && end != std::string::npos;
       start = end + 1) {
    std::string token;
-    status = OptionTypeInfo::NextToken(value, separator, start, &end, &token);
+    status =
+        OptionTypeInfo::NextToken(stripped, separator, start, &end, &token);
    if (status.ok()) {
      T elem;
      status = elem_info.Parse(copy, name, token, &elem);
@@ -557,11 +557,13 @@ static std::unordered_map<std::string, OptionTypeInfo>
            embedded.ignore_unsupported_options = true;
            std::vector<std::shared_ptr<EventListener>> listeners;
            Status s;
+            // Permit `{item1:item2}` or bare `item1:item2`.
+            std::string stripped = OptionTypeInfo::StripOuterBraces(value);
            for (size_t start = 0, end = 0;
-                 s.ok() && start < value.size() && end != std::string::npos;
+                 s.ok() && start < stripped.size() && end != std::string::npos;
                 start = end + 1) {
              std::string token;
-              s = OptionTypeInfo::NextToken(value, ':', start, &end, &token);
+              s = OptionTypeInfo::NextToken(stripped, ':', start, &end, &token);
              if (s.ok() && !token.empty()) {
                std::shared_ptr<EventListener> listener;
                s = EventListener::CreateFromString(embedded, token, &listener);
@@ -770,6 +770,11 @@ Status StringToMap(const std::string& opts_str,
  // Example:
  //   opts_str = "write_buffer_size=1024;max_write_buffer_number=2;"
  //              "nested_opt={opt1=1;opt2=2};max_bytes_for_level_base=100"
+  //
+  // Each value in the resulting map can be substituted directly into a
+  // `key=value;` context (e.g. SetOptions) and re-parsed without ambiguity:
+  // values that were wrapped in `{...}` in the input keep their wrapping,
+  // so embedded `;` characters stay escaped.
  size_t pos = 0;
  std::string opts = trim(opts_str);
  // If the input string starts and ends with "{...}", strip off the brackets
@@ -790,23 +795,78 @@ Status StringToMap(const std::string& opts_str,
      return Status::InvalidArgument("Empty key found");
    }

+    // Locate value start (after '=' and any whitespace).
+    size_t value_start = eq_pos + 1;
+    while (value_start < opts.size() && isspace(opts[value_start])) {
+      ++value_start;
+    }
+
    std::string value;
-    Status s = OptionTypeInfo::NextToken(opts, ';', eq_pos + 1, &pos, &value);
-    if (!s.ok()) {
-      return s;
-    } else {
-      (*opts_map)[key] = value;
-      if (pos == std::string::npos) {
-        break;
-      } else {
-        pos++;
+    if (value_start < opts.size() && opts[value_start] == '{') {
+      // Braced value: extract the entire `{...}` substring INCLUDING braces
+      // so that the value can be re-emitted as-is in a `key=value;` context.
+      int count = 1;
+      size_t brace_pos = value_start + 1;
+      while (brace_pos < opts.size() && count > 0) {
+        if (opts[brace_pos] == '{') {
+          ++count;
+        } else if (opts[brace_pos] == '}') {
+          --count;
+        }
+        if (count > 0) {
+          ++brace_pos;
+        }
      }
+      if (count != 0) {
+        return Status::InvalidArgument(
+            "Mismatched curly braces for nested options");
+      }
+      value = opts.substr(value_start, brace_pos - value_start + 1);
+      pos = brace_pos + 1;
+      // Allow whitespace, then either delimiter or end.
+      while (pos < opts.size() && isspace(opts[pos])) {
+        ++pos;
+      }
+      if (pos < opts.size() && opts[pos] != ';') {
+        return Status::InvalidArgument("Unexpected chars after nested options");
+      }
+    } else {
+      // Non-braced: scan to the next ';'. The value may contain '=' (only
+      // the first '=' separates key from value).
+      size_t end = opts.find(';', value_start);
+      if (end == std::string::npos) {
+        value = trim(opts.substr(value_start));
+        pos = std::string::npos;
+      } else {
+        value = trim(opts.substr(value_start, end - value_start));
+        pos = end;
+      }
+    }
+
+    (*opts_map)[key] = value;
+    if (pos == std::string::npos) {
+      break;
+    } else {
+      pos++;
    }
  }

  return Status::OK();
 }

+Status MapToString(const std::unordered_map<std::string, std::string>& opts_map,
+                   std::string* opts_str) {
+  assert(opts_str);
+  opts_str->clear();
+  for (const auto& [key, value] : opts_map) {
+    opts_str->append(key);
+    opts_str->append("=");
+    opts_str->append(value);
+    opts_str->append(";");
+  }
+  return Status::OK();
+}
+
 Status GetStringFromDBOptions(std::string* opt_string,
                              const DBOptions& db_options,
                              const std::string& delimiter) {
@@ -1053,6 +1113,32 @@ Status OptionTypeInfo::NextToken(const std::string& opts, char delimiter,
  return Status::OK();
 }

+std::string OptionTypeInfo::StripOuterBraces(const std::string& value) {
+  if (value.size() < 2 || value.front() != '{' || value.back() != '}') {
+    return value;
+  }
+  // Verify the leading '{' actually pairs with the trailing '}'.
+  // For `{a:b}` the leading brace closes only at the trailing `}` (one
+  // matching pair, OK to strip). For `{a}:{b}` the leading brace closes
+  // at the first inner `}`, so the leading and trailing braces are
+  // independent -- leave the value alone. We strip at most one layer:
+  // each layer of `{}` corresponds to one level of nesting that the
+  // encoder added for that level, and the typed parser at this level
+  // wants to peel exactly its own layer.
+  int depth = 0;
+  for (size_t i = 0; i + 1 < value.size(); ++i) {
+    if (value[i] == '{') {
+      ++depth;
+    } else if (value[i] == '}') {
+      --depth;
+      if (depth == 0) {
+        return value;  // outer braces are independent
+      }
+    }
+  }
+  return value.substr(1, value.size() - 2);
+}
+
 Status OptionTypeInfo::Parse(const ConfigOptions& config_options,
                             const std::string& opt_name,
                             const std::string& value, void* opt_ptr) const {
@@ -1071,7 +1157,13 @@ Status OptionTypeInfo::Parse(const ConfigOptions& config_options,
      copy.invoke_prepare_options = false;
      void* opt_addr = GetOffset(opt_ptr);
      return parse_func_(copy, opt_name, opt_value, opt_addr);
-    } else if (ParseOptionHelper(GetOffset(opt_ptr), type_, opt_value)) {
+    } else if (ParseOptionHelper(GetOffset(opt_ptr), type_,
+                                 StripOuterBraces(opt_value))) {
+      // Scalar types (int, bool, enum, string, ...). StringToMap now
+      // preserves outer braces in nested values, so a hand-crafted
+      // `key={42}` (or `db_log_dir={/tmp}`) lands here with a wrap
+      // that scalar parsers don't expect; permissively strip one
+      // level so braced scalar input still parses as before.
      return Status::OK();
    } else if (IsConfigurable()) {
      // The option is <config>.<name>
@@ -1938,11 +1938,13 @@ TEST_F(OptionsTest, StringToMapTest) {
  ASSERT_EQ(opts_map["k2"], "");
  ASSERT_TRUE(opts_map.find("k3") != opts_map.end());
  ASSERT_EQ(opts_map["k3"], "");
-  // Regular nested options
+  // Regular nested options. Braces are preserved on values that came in
+  // braced, so the value is self-contained for direct embedding via
+  // `key=value;` and round-trips through StringToMap/MapToString.
  opts_map.clear();
  ASSERT_OK(StringToMap("k1=v1;k2={nk1=nv1;nk2=nv2};k3=v3", &opts_map));
  ASSERT_EQ(opts_map["k1"], "v1");
-  ASSERT_EQ(opts_map["k2"], "nk1=nv1;nk2=nv2");
+  ASSERT_EQ(opts_map["k2"], "{nk1=nv1;nk2=nv2}");
  ASSERT_EQ(opts_map["k3"], "v3");
  // Multi-level nested options
  opts_map.clear();
@@ -1951,34 +1953,35 @@ TEST_F(OptionsTest, StringToMapTest) {
                  "k3={nk1={nnk1={nnnk1=nnnv1;nnnk2;nnnv2}}};k4=v4",
                  &opts_map));
  ASSERT_EQ(opts_map["k1"], "v1");
-  ASSERT_EQ(opts_map["k2"], "nk1=nv1;nk2={nnk1=nnk2}");
-  ASSERT_EQ(opts_map["k3"], "nk1={nnk1={nnnk1=nnnv1;nnnk2;nnnv2}}");
+  ASSERT_EQ(opts_map["k2"], "{nk1=nv1;nk2={nnk1=nnk2}}");
+  ASSERT_EQ(opts_map["k3"], "{nk1={nnk1={nnnk1=nnnv1;nnnk2;nnnv2}}}");
  ASSERT_EQ(opts_map["k4"], "v4");
  // Garbage inside curly braces
  opts_map.clear();
  ASSERT_OK(StringToMap("k1=v1;k2={dfad=};k3={=};k4=v4", &opts_map));
  ASSERT_EQ(opts_map["k1"], "v1");
-  ASSERT_EQ(opts_map["k2"], "dfad=");
-  ASSERT_EQ(opts_map["k3"], "=");
+  ASSERT_EQ(opts_map["k2"], "{dfad=}");
+  ASSERT_EQ(opts_map["k3"], "{=}");
  ASSERT_EQ(opts_map["k4"], "v4");
  // Empty nested options
  opts_map.clear();
  ASSERT_OK(StringToMap("k1=v1;k2={};", &opts_map));
  ASSERT_EQ(opts_map["k1"], "v1");
-  ASSERT_EQ(opts_map["k2"], "");
+  ASSERT_EQ(opts_map["k2"], "{}");
  opts_map.clear();
  ASSERT_OK(StringToMap("k1=v1;k2={{{{}}}{}{}};", &opts_map));
  ASSERT_EQ(opts_map["k1"], "v1");
-  ASSERT_EQ(opts_map["k2"], "{{{}}}{}{}");
-  // With random spaces
+  ASSERT_EQ(opts_map["k2"], "{{{{}}}{}{}}");
+  // With random spaces. Internal whitespace is preserved verbatim now
+  // (we keep the original substring of the value rather than trimming).
  opts_map.clear();
  ASSERT_OK(
      StringToMap("  k1 =  v1 ; k2= {nk1=nv1; nk2={nnk1=nnk2}}  ; "
                  "k3={  {   } }; k4= v4  ",
                  &opts_map));
  ASSERT_EQ(opts_map["k1"], "v1");
-  ASSERT_EQ(opts_map["k2"], "nk1=nv1; nk2={nnk1=nnk2}");
-  ASSERT_EQ(opts_map["k3"], "{   }");
+  ASSERT_EQ(opts_map["k2"], "{nk1=nv1; nk2={nnk1=nnk2}}");
+  ASSERT_EQ(opts_map["k3"], "{  {   } }");
  ASSERT_EQ(opts_map["k4"], "v4");

  // Empty key
@@ -2008,6 +2011,165 @@ TEST_F(OptionsTest, StringToMapTest) {
  ASSERT_NOK(StringToMap("k1=v1;k2={{dfdl}adfa}{}", &opts_map));
 }

+TEST_F(OptionsTest, EmptyBracedVectorAndBracedScalarTest) {
+  // An empty braced list value (`key={}`) must parse as an empty
+  // collection, not as a vector with one empty element. StringToMap
+  // preserves the braces, so the typed vector parser sees `{}`; its
+  // outer-brace strip needs to handle the size-2 case.
+  ColumnFamilyOptions cf_opts;
+  ConfigOptions cfg;
+  ASSERT_OK(GetColumnFamilyOptionsFromString(
+      cfg, ColumnFamilyOptions(), "compression_per_level={};", &cf_opts));
+  EXPECT_TRUE(cf_opts.compression_per_level.empty());
+
+  // A scalar value wrapped in `{}` (e.g. hand-crafted `key={42}`) must
+  // still parse. StringToMap preserves the braces, so the scalar
+  // dispatch in OptionTypeInfo::Parse permissively strips one outer
+  // layer.
+  DBOptions db_opts;
+  ASSERT_OK(GetDBOptionsFromString(cfg, DBOptions(), "max_open_files={42};",
+                                   &db_opts));
+  EXPECT_EQ(db_opts.max_open_files, 42);
+
+  ASSERT_OK(GetDBOptionsFromString(cfg, DBOptions(),
+                                   "create_if_missing={true};", &db_opts));
+  EXPECT_TRUE(db_opts.create_if_missing);
+}
+
+TEST_F(OptionsTest, MapToStringTest) {
+  using Map = std::unordered_map<std::string, std::string>;
+  std::string s;
+
+  // Simple values.
+  Map simple{{"a", "1"}, {"b", "hello"}};
+  ASSERT_OK(MapToString(simple, &s));
+  Map reparsed;
+  ASSERT_OK(StringToMap(s, &reparsed));
+  ASSERT_EQ(reparsed, simple);
+
+  // A value already in self-contained braced form (as returned by
+  // StringToMap) passes through unchanged. No extra wrapping is needed
+  // for round-trip.
+  Map already_braced{{"k", "{a=b;c=d}"}};
+  ASSERT_OK(MapToString(already_braced, &s));
+  ASSERT_EQ(s, "k={a=b;c=d};");
+  Map reparsed_braced;
+  ASSERT_OK(StringToMap(s, &reparsed_braced));
+  ASSERT_EQ(reparsed_braced, already_braced);
+}
+
+TEST_F(OptionsTest, StringToMapMapToStringRoundTripTest) {
+  using Map = std::unordered_map<std::string, std::string>;
+  // Pull a single entry out of a StringToMap-ed map and re-embed it
+  // directly into a `key=value;` SetOptions-style string. The pulled
+  // value must round-trip without any extra escaping by the caller --
+  // this is the property that lets users compose option strings from
+  // map entries safely.
+  const std::string original =
+      "filter_policy={id=ribbonfilter:10:-1;bloom_before_level=-1;};"
+      "block_size=4096;cache_index_and_filter_blocks=true";
+
+  Map parsed;
+  ASSERT_OK(StringToMap(original, &parsed));
+  // The filter_policy value retains its braces -- it's self-contained.
+  EXPECT_EQ(parsed["filter_policy"],
+            "{id=ribbonfilter:10:-1;bloom_before_level=-1;}");
+  EXPECT_EQ(parsed["block_size"], "4096");
+  EXPECT_EQ(parsed["cache_index_and_filter_blocks"], "true");
+
+  // Pull a single entry out and embed it in a fresh `key=value;`
+  // string. No extra wrapping required by the caller.
+  const std::string embedded =
+      "filter_policy=" + parsed["filter_policy"] + ";other_opt=42";
+  Map reparsed;
+  ASSERT_OK(StringToMap(embedded, &reparsed));
+  EXPECT_EQ(reparsed["filter_policy"], parsed["filter_policy"]);
+  EXPECT_EQ(reparsed["other_opt"], "42");
+
+  // Whole-map round-trip via MapToString.
+  std::string regenerated;
+  ASSERT_OK(MapToString(parsed, &regenerated));
+  Map reparsed2;
+  ASSERT_OK(StringToMap(regenerated, &reparsed2));
+  EXPECT_EQ(reparsed2, parsed);
+}
+
+TEST_F(OptionsTest, FullOptionsStringToMapRoundTripTest) {
+  // Round-trip a populated DBOptions and CFOptions through the
+  // StringToMap -> MapToString path. This catches any custom option
+  // serializer that emits a value containing the StringToMap delimiter
+  // (';') without enclosing it in '{}' -- such a value would survive
+  // StringToMap as a self-contained entry but corrupt subsequent
+  // re-serialization. Existing typed tests
+  // (ColumnFamilyOptionsSerialization, DBOptionsSerialization) cover the
+  // typed Configurable round-trip; this test specifically exercises the
+  // public StringToMap + MapToString path that callers use to compose,
+  // mutate, and re-emit option strings.
+  Random rnd(607);
+  ConfigOptions config_options;
+  config_options.input_strings_escaped = false;
+
+  // ---- ColumnFamilyOptions ----
+  Options options;
+  ColumnFamilyOptions base_cf_opt;
+  base_cf_opt.comparator = test::BytewiseComparatorWithU64TsWrapper();
+  test::RandomInitCFOptions(&base_cf_opt, options, &rnd);
+  std::string cf_str;
+  ASSERT_OK(
+      GetStringFromColumnFamilyOptions(config_options, base_cf_opt, &cf_str));
+
+  std::unordered_map<std::string, std::string> cf_map;
+  ASSERT_OK(StringToMap(cf_str, &cf_map));
+  // Each entry must be embeddable as-is in a `key=value;` context.
+  for (const auto& [k, v] : cf_map) {
+    std::string solo;
+    solo.append(k).append("=").append(v).append(";");
+    std::unordered_map<std::string, std::string> solo_map;
+    ASSERT_OK(StringToMap(solo, &solo_map))
+        << "single-entry round-trip failed for " << k << ": " << v;
+    ASSERT_EQ(solo_map.size(), 1u) << k << "=" << v;
+    ASSERT_EQ(solo_map[k], v) << "single-entry value mismatch for " << k;
+  }
+  // Whole-map MapToString must reproduce a string parseable into a CF
+  // options object equivalent to the original.
+  std::string cf_round;
+  ASSERT_OK(MapToString(cf_map, &cf_round));
+  ColumnFamilyOptions cf_back;
+  ASSERT_OK(GetColumnFamilyOptionsFromString(
+      config_options, ColumnFamilyOptions(), cf_round, &cf_back));
+  ASSERT_OK(RocksDBOptionsParser::VerifyCFOptions(config_options, base_cf_opt,
+                                                  cf_back));
+
+  // ---- DBOptions ----
+  DBOptions base_db_opt;
+  test::RandomInitDBOptions(&base_db_opt, &rnd);
+  std::string db_str;
+  ASSERT_OK(GetStringFromDBOptions(config_options, base_db_opt, &db_str));
+
+  std::unordered_map<std::string, std::string> db_map;
+  ASSERT_OK(StringToMap(db_str, &db_map));
+  for (const auto& [k, v] : db_map) {
+    std::string solo;
+    solo.append(k).append("=").append(v).append(";");
+    std::unordered_map<std::string, std::string> solo_map;
+    ASSERT_OK(StringToMap(solo, &solo_map))
+        << "single-entry round-trip failed for " << k << ": " << v;
+    ASSERT_EQ(solo_map.size(), 1u) << k << "=" << v;
+    ASSERT_EQ(solo_map[k], v) << "single-entry value mismatch for " << k;
+  }
+  std::string db_round;
+  ASSERT_OK(MapToString(db_map, &db_round));
+  DBOptions db_back;
+  ASSERT_OK(
+      GetDBOptionsFromString(config_options, DBOptions(), db_round, &db_back));
+  ASSERT_OK(RocksDBOptionsParser::VerifyDBOptions(config_options, base_db_opt,
+                                                  db_back));
+
+  // RandomInitCFOptions allocates a raw CompactionFilter*; the
+  // ColumnFamilyOptions destructor doesn't own it.
+  delete base_cf_opt.compaction_filter;
+}
+
 TEST_F(OptionsTest, StringToMapRandomTest) {
  std::unordered_map<std::string, std::string> opts_map;
  // Make sure segfault is not hit by semi-random strings
@@ -2028,6 +2190,15 @@ TEST_F(OptionsTest, StringToMapRandomTest) {
        str[pos] = ' ';
        Status s = StringToMap(str, &opts_map);
        ASSERT_TRUE(s.ok() || s.IsInvalidArgument());
+        if (s.ok()) {
+          // Round-trip: StringToMap entries are self-contained, so
+          // MapToString of them must parse back to the same map.
+          std::string regen;
+          ASSERT_OK(MapToString(opts_map, &regen));
+          std::unordered_map<std::string, std::string> reparsed;
+          ASSERT_OK(StringToMap(regen, &reparsed));
+          ASSERT_EQ(reparsed, opts_map) << "regen=" << regen;
+        }
        opts_map.clear();
      }
    }
@@ -2047,8 +2218,22 @@ TEST_F(OptionsTest, StringToMapRandomTest) {
    }
    Status s = StringToMap(str, &opts_map);
    ASSERT_TRUE(s.ok() || s.IsInvalidArgument());
+    if (s.ok()) {
+      std::string regen;
+      ASSERT_OK(MapToString(opts_map, &regen));
+      std::unordered_map<std::string, std::string> reparsed;
+      ASSERT_OK(StringToMap(regen, &reparsed));
+      ASSERT_EQ(reparsed, opts_map) << "regen=" << regen;
+    }
    s = StringToMap("name=" + str, &opts_map);
    ASSERT_TRUE(s.ok() || s.IsInvalidArgument());
+    if (s.ok()) {
+      std::string regen;
+      ASSERT_OK(MapToString(opts_map, &regen));
+      std::unordered_map<std::string, std::string> reparsed;
+      ASSERT_OK(StringToMap(regen, &reparsed));
+      ASSERT_EQ(reparsed, opts_map) << "regen=" << regen;
+    }
    opts_map.clear();
  }
 }
@@ -506,6 +506,7 @@ TEST_MAIN_SOURCES =                                                     \
  db/db_clip_test.cc                                                    \
  db/db_dynamic_level_test.cc                                           \
  db/db_encryption_test.cc                                              \
+  db/db_etc2_test.cc                                                    \
  db/db_etc3_test.cc                                                    \
  db/db_flush_test.cc                                                   \
  db/db_follower_test.cc						                                    \
@@ -533,7 +534,6 @@ TEST_MAIN_SOURCES =                                                     \
  db/db_table_properties_test.cc                                        \
  db/db_tailing_iter_test.cc                                            \
  db/db_test.cc                                                         \
-  db/db_test2.cc                                                        \
  db/db_logical_block_size_cache_test.cc                                \
  db/db_universal_compaction_test.cc                                    \
  db/db_wal_test.cc                                                     \
@@ -1077,6 +1077,9 @@ void BlockBasedTableIterator::Prepare(const MultiScanArgs* multiscan_opts) {
      multiscan_opts->io_coalesce_threshold;
  job->job_options.read_options = read_options_;
  job->job_options.read_options.async_io = multiscan_opts->use_async_io;
+  // CollectBlockHandles walks sorted scan ranges through the table index, whose
+  // key order matches data block offset order.
+  job->job_options.block_handles_are_sorted = true;

  std::shared_ptr<ReadSet> read_set;
  // IODispatcher should be provided by DBIter::Prepare() to enable sharing
@@ -67,6 +67,8 @@ scriptpath=`dirname ${BASH_SOURCE[0]}`

 test_dir=${TEST_TMPDIR:-"/tmp"}"/rocksdb_format_compatible_$USER"
 rm -rf ${test_dir:?}
+# Also clean the noexec fallback dir from any prior run (see cs_test_dir below).
+rm -rf "$PWD/tmp/cs"

 # Prevent 'make clean' etc. from wiping out test_dir
 export TEST_TMPDIR=$test_dir"/misc"
@@ -88,6 +90,17 @@ mkdir -p $db_test_dir
 # For backup/restore test (uses DB test)
 bak_test_dir=$test_dir"/bak"
 mkdir -p $bak_test_dir
+# For remote compaction test.  The saved ldb binary must be executable.
+# Detect noexec early and fall back to $PWD/tmp/cs (git-ignored).
+cs_test_dir=$test_dir"/cs"
+mkdir -p $cs_test_dir
+if cp /bin/true $cs_test_dir/_exec_test 2>/dev/null && \
+   ! $cs_test_dir/_exec_test 2>/dev/null; then
+  echo "   $cs_test_dir is noexec, using $PWD/tmp/cs instead"
+  cs_test_dir="$PWD/tmp/cs"
+  mkdir -p $cs_test_dir
+fi
+rm -f $cs_test_dir/_exec_test

 python_bin=$(which python3 || which python || echo python3)

@@ -238,7 +251,7 @@ invoke_make()
 generate_db()
 {
    set +e
-    [ "$SANITY_CHECK" ] || bash "$script_copy_dir"/generate_random_db.sh "$1" "$2"
+    [ "$SANITY_CHECK" ] || bash "$script_copy_dir"/generate_random_db.sh "$1" "$2" "$3"
    if [ $? -ne 0 ]; then
        echo ==== Error loading data from $2 to $1 ====
        exit 1
@@ -320,6 +333,40 @@ member_of_array()
  return 1
 }

+# Run one cross-version remote compaction compatibility test.
+# Generates a DB using the primary's ldb (so its OPTIONS file matches the
+# primary's version), then runs primary and worker from potentially
+# different versions. They coordinate via files in test_dir (see
+# remote_compaction_primary/worker ldb commands for protocol details).
+# Tests the wire format of CompactionServiceInput/Result between versions.
+run_cs_compat_test()
+{
+  local test_label="$1"
+  local primary_ldb="$2"
+  local worker_ldb="$3"
+  local cs_run_dir="$4"
+
+  echo "== $test_label"
+  rm -rf "$cs_run_dir" && mkdir -p "$cs_run_dir"
+  generate_db $input_data_path "$cs_run_dir/db" "$primary_ldb"
+  # Write keys spanning the full key range to create an L0 file that
+  # overlaps with existing L0 files, ensuring CompactRange triggers a
+  # real compaction (not a trivial move).
+  printf "a ==> overlap_start\nzzzzzzz ==> overlap_end\n" | \
+    $primary_ldb load --db="$cs_run_dir/db" --auto_compaction=false
+  $worker_ldb remote_compaction_worker --db="$cs_run_dir/db" --job_dir="$cs_run_dir" &
+  local worker_pid=$!
+  local primary_exit=0
+  $primary_ldb remote_compaction_primary --db="$cs_run_dir/db" --job_dir="$cs_run_dir" || primary_exit=$?
+  local worker_exit=0
+  wait $worker_pid || worker_exit=$?
+  if [ $primary_exit -ne 0 ] || [ $worker_exit -ne 0 ]; then
+    echo "==== Error running remote compaction: $test_label (primary=$primary_exit worker=$worker_exit) ===="
+    kill $worker_pid 2>/dev/null || true
+    exit 1
+  fi
+}
+
 force_no_fbcode()
 {
  # Not all branches recognize ROCKSDB_NO_FBCODE and we should not need
@@ -339,8 +386,10 @@ force_no_fbcode()
 # * (Again) check out, build, and do (other) stuff with the "current"
 #    branch, potentially using data from older branches.
 #
-# This way, we only do at most n+1 checkout+build steps, without the
-# need to stash away executables.
+# This way, we only do at most n+1 checkout+build steps.  The one
+# exception is the remote compaction test, which saves a copy of the
+# current ldb binary before old-ref checkouts overwrite ./ldb, so both
+# versions can run simultaneously as primary and worker.

 # Decorate name
 current_checkout_name="$current_checkout_name ($current_checkout_hash)"
@@ -351,6 +400,33 @@ force_no_fbcode
 invoke_make clean
 DISABLE_WARNING_AS_ERROR=1 invoke_make ldb -j$J

+# Save current ldb for cross-version remote compaction tests.  The old-ref
+# checkout will overwrite ./ldb, so we save a copy now.
+save_cs_current_ldb()
+{
+  cs_current_ldb=$cs_test_dir/current_ldb
+  cp -f ./ldb $cs_current_ldb
+  # Copy shared libs next to the binary for LD_LIBRARY_PATH.
+  # Static builds have no .so files - cp fails harmlessly.
+  cp -f ./librocksdb*.so* $cs_test_dir/ 2>/dev/null || true
+  cs_current_ldb_cmd="env LD_LIBRARY_PATH=$cs_test_dir $cs_current_ldb"
+  cs_ldb_output=$($cs_current_ldb_cmd --version 2>&1)
+  cs_ldb_exit=$?
+  return $cs_ldb_exit
+}
+echo "== Saving current ldb for remote compaction cross-version tests"
+cs_current_ldb_cmd=""
+if [ ! "$SANITY_CHECK" ]; then
+  if ! save_cs_current_ldb; then
+    echo "==== Error: saved current ldb cannot run from $cs_test_dir (exit=$cs_ldb_exit): $cs_ldb_output ===="
+    exit 1
+  fi
+  # Smoke test before cross-version tests to catch integration failures early.
+  run_cs_compat_test \
+    "Remote compaction smoke test: current primary + current worker" \
+    "$cs_current_ldb_cmd" "$cs_current_ldb_cmd" "$cs_test_dir/smoke"
+fi
+
 echo "== Using $current_checkout_name, generate DB with extern SST and ingest"
 current_ext_test_dir=$ext_test_dir"/current"
 write_external_sst $input_data_path ${current_ext_test_dir}_pointless $current_ext_test_dir
@@ -429,6 +505,30 @@ do
    restore_db $current_bak_test_dir $db_test_dir/$checkout_ref
    compare_db $db_test_dir/$checkout_ref $current_db_test_dir forward_${checkout_ref}_dump.txt 0
  fi
+
+  # Remote compaction format compatibility: test that primary and worker from
+  # different versions can exchange CompactionServiceInput/Result.
+  # Requires db_forward_with_options_refs (the old worker must be able to
+  # read OPTIONS written by the current primary) and the remote compaction
+  # related ldb commands.
+  # Skipped in SANITY_CHECK mode (no ldb binary to test).
+  if [ ! "$SANITY_CHECK" ] &&
+    [ -n "$cs_current_ldb_cmd" ] &&
+    member_of_array "$checkout_ref" "${db_forward_with_options_refs[@]}"
+  then
+    if ./ldb --help 2>&1 | grep -q remote_compaction_primary; then
+      cs_old_ldb_cmd="./ldb"
+      ref_dir=$cs_test_dir/$checkout_ref
+      run_cs_compat_test \
+        "Remote compaction compatibility: current primary + $checkout_ref worker" \
+        "$cs_current_ldb_cmd" "$cs_old_ldb_cmd" "$ref_dir/test1"
+      run_cs_compat_test \
+        "Remote compaction compatibility: $checkout_ref primary + current worker" \
+        "$cs_old_ldb_cmd" "$cs_current_ldb_cmd" "$ref_dir/test2"
+    else
+      echo "   remote_compaction commands not available at $checkout_ref, skipping"
+    fi
+  fi
 done

 echo "== Building $current_checkout_name debug (again, final)"
@@ -466,6 +566,7 @@ do
    restore_db $bak_test_dir/$checkout_ref $db_test_dir/$checkout_ref
    compare_db $db_test_dir/$checkout_ref $current_db_test_dir db_dump.txt 1 0
  fi
+
 done

 if [ "$SANITY_CHECK" ]; then
@@ -1,28 +1,29 @@
 #!/usr/bin/env bash
 # Copyright (c) Facebook, Inc. and its affiliates. All Rights Reserved.
 #
-# A shell script to load some pre generated data file to a DB using ldb tool
-# ./ldb needs to be avaible to be executed.
+# A shell script to load some pre generated data file to a DB using ldb tool.
+# <ldb_command> must be available to execute (defaults to ./ldb).
 #
-# Usage: <SCRIPT> <input_data_path> <DB Path>
+# Usage: <SCRIPT> <input_data_path> <DB Path> [<ldb_command>]

 if [ "$#" -lt 2 ]; then
-  echo "usage: $BASH_SOURCE <input_data_path> <DB Path>"
+  echo "usage: $BASH_SOURCE <input_data_path> <DB Path> [<ldb_command>]"
  exit 1
 fi

 input_data_dir=$1
 db_dir=$2
+ldb_cmd=${3:-./ldb}
 rm -rf $db_dir

 second_gen_compression_support=
 mixed_compression_support=
 # Support for `ldb --version` is a crude under-approximation for versions
 # supporting dictionary compression and algorithms including zstd and lz4
-if ./ldb --version 2>/dev/null >/dev/null; then
+if $ldb_cmd --version 2>/dev/null >/dev/null; then
  second_gen_compression_support=1

-  if ./ldb load --db=$db_dir --compression_type=mixed --create_if_missing \
+  if $ldb_cmd load --db=$db_dir --compression_type=mixed --create_if_missing \
      < /dev/null 2>/dev/null >/dev/null; then
    mixed_compression_support=1
  fi
@@ -31,7 +32,7 @@ fi

 # Check if deleterange command is supported by grepping ldb --help
 deleterange_support=
-if ./ldb --help 2>&1 | grep -q deleterange; then
+if $ldb_cmd --help 2>&1 | grep -q deleterange; then
  deleterange_support=1
 fi

@@ -69,7 +70,7 @@ do
  else
    d_arg=""
  fi
-  ./ldb load --db=$db_dir --compression_type=$c $d_arg --bloom_bits=10 \
+  $ldb_cmd load --db=$db_dir --compression_type=$c $d_arg --bloom_bits=10 \
    --auto_compaction=false --create_if_missing < $input_data_dir/$f

  # Use md5sum of file to deterministically decide whether to add a range
@@ -89,11 +90,11 @@ do
        end_key="${key}0"
        if [ "$deleterange_support" == "1" ]; then
          echo "== Deleting range [$key, $end_key) from $f"
-          ./ldb deleterange --db=$db_dir "$key" "$end_key"
+          $ldb_cmd deleterange --db=$db_dir "$key" "$end_key"
        else
          # Fall back to point delete for equivalent logical contents
          echo "== Deleting key $key from $f"
-          ./ldb delete --db=$db_dir "$key"
+          $ldb_cmd delete --db=$db_dir "$key"
        fi
      fi
    fi
@@ -7,6 +7,7 @@
 #include "rocksdb/utilities/ldb_cmd.h"

 #include <cstddef>
+#include <cstdio>
 #include <cstdlib>
 #include <ctime>
 #include <fstream>
@@ -16,6 +17,7 @@
 #include <sstream>
 #include <stdexcept>
 #include <string>
+#include <thread>

 #include "db/blob/blob_index.h"
 #include "db/db_impl/db_impl.h"
@@ -434,6 +436,14 @@ LDBCommand* LDBCommand::SelectCommand(const ParsedParams& parsed_params) {
    return new CompactionProgressDumpCommand(parsed_params.cmd_params,
                                             parsed_params.option_map,
                                             parsed_params.flags);
+  } else if (parsed_params.cmd == RemoteCompactionPrimaryCommand::Name()) {
+    return new RemoteCompactionPrimaryCommand(parsed_params.cmd_params,
+                                              parsed_params.option_map,
+                                              parsed_params.flags);
+  } else if (parsed_params.cmd == RemoteCompactionWorkerCommand::Name()) {
+    return new RemoteCompactionWorkerCommand(parsed_params.cmd_params,
+                                             parsed_params.option_map,
+                                             parsed_params.flags);
  }
  return nullptr;
 }
@@ -5454,4 +5464,191 @@ void CompactionProgressDumpCommand::DoCommand() {
  DumpCompactionProgressFile(path_);
 }

+namespace {
+
+const std::string kInputFile = "input.bin";
+const std::string kResultFile = "result.bin";
+constexpr int kPollIntervalMs = 100;
+constexpr int kPollTimeoutMs = 120000;
+constexpr int kPollMaxAttempts = kPollTimeoutMs / kPollIntervalMs;
+
+// Atomic write via tmp+rename so the polling reader never sees partial data.
+bool AtomicWriteStringToFile(const std::string& path, const std::string& data) {
+  std::string tmp_path = path + ".tmp";
+  std::ofstream ofs(tmp_path, std::ios::binary);
+  if (!ofs) {
+    return false;
+  }
+  ofs.write(data.data(), data.size());
+  if (!ofs.good()) {
+    return false;
+  }
+  ofs.close();
+  return std::rename(tmp_path.c_str(), path.c_str()) == 0;
+}
+
+bool PollForFile(Env* env, const std::string& path, std::string* data) {
+  for (int i = 0; i < kPollMaxAttempts; i++) {
+    Status s = ROCKSDB_NAMESPACE::ReadFileToString(env, path, data);
+    if (s.ok() && !data->empty()) {
+      return true;
+    }
+    std::this_thread::sleep_for(std::chrono::milliseconds(kPollIntervalMs));
+  }
+  return false;
+}
+
+class LocalFileCompactionService : public CompactionService {
+ public:
+  explicit LocalFileCompactionService(const std::string& job_dir)
+      : job_dir_(job_dir) {}
+
+  const char* Name() const override { return "LocalFileCompactionService"; }
+
+  CompactionServiceScheduleResponse Schedule(
+      const CompactionServiceJobInfo& /*info*/,
+      const std::string& compaction_service_input) override {
+    assert(!scheduled_);
+    scheduled_ = true;
+    if (!AtomicWriteStringToFile(job_dir_ + "/" + kInputFile,
+                                 compaction_service_input)) {
+      return CompactionServiceScheduleResponse(
+          CompactionServiceJobStatus::kFailure);
+    }
+    return CompactionServiceScheduleResponse(
+        "job_0001", CompactionServiceJobStatus::kSuccess);
+  }
+
+  CompactionServiceJobStatus Wait(const std::string& /*scheduled_job_id*/,
+                                  std::string* result) override {
+    if (!PollForFile(Env::Default(), job_dir_ + "/" + kResultFile, result)) {
+      return CompactionServiceJobStatus::kFailure;
+    }
+    return CompactionServiceJobStatus::kSuccess;
+  }
+
+ private:
+  std::string job_dir_;
+  bool scheduled_ = false;
+};
+
+}  // namespace
+
+const std::string RemoteCompactionPrimaryCommand::ARG_JOB_DIR =
+    "job_dir";  // NOLINT(cert-err58-cpp)
+
+RemoteCompactionPrimaryCommand::RemoteCompactionPrimaryCommand(
+    const std::vector<std::string>& /*params*/,
+    const std::map<std::string, std::string>& options,
+    const std::vector<std::string>& flags)
+    : LDBCommand(options, flags, false, BuildCmdLineOptions({ARG_JOB_DIR})) {
+  auto it = options.find(ARG_JOB_DIR);
+  if (it != options.end()) {
+    job_dir_ = it->second;
+  } else {
+    exec_state_ = LDBCommandExecuteResult::Failed("--job_dir is required");
+  }
+}
+
+void RemoteCompactionPrimaryCommand::Help(std::string& ret) {
+  ret.append("  ");
+  ret.append(Name());
+  ret.append(" --db=<db_path> --job_dir=<dir>");
+  ret.append("\n    Open an existing DB and run CompactRange() through ");
+  ret.append("LocalFileCompactionService.\n");
+  ret.append("    Schedule() writes job_dir/input.bin (once). ");
+  ret.append("Wait() polls for job_dir/result.bin.\n");
+  ret.append("    Must run with remote_compaction_worker in background on ");
+  ret.append("the same --db and --job_dir,\n");
+  ret.append("    otherwise Wait() will hang until timeout.\n");
+}
+
+void RemoteCompactionPrimaryCommand::DoCommand() {
+  auto cs = std::make_shared<LocalFileCompactionService>(job_dir_);
+
+  Options options;
+  options.compaction_service = cs;
+  options.disable_auto_compactions = true;
+
+  std::unique_ptr<DB> db;
+  Status s = DB::Open(options, db_path_, &db);
+  if (!s.ok()) {
+    exec_state_ = LDBCommandExecuteResult::Failed("Open: " + s.ToString());
+    return;
+  }
+
+  s = db->CompactRange(CompactRangeOptions(), nullptr, nullptr);
+  if (!s.ok()) {
+    exec_state_ =
+        LDBCommandExecuteResult::Failed("CompactRange: " + s.ToString());
+    return;
+  }
+
+  fprintf(stdout, "remote_compaction_primary: OK\n");
+}
+
+const std::string RemoteCompactionWorkerCommand::ARG_JOB_DIR =
+    "job_dir";  // NOLINT(cert-err58-cpp)
+
+RemoteCompactionWorkerCommand::RemoteCompactionWorkerCommand(
+    const std::vector<std::string>& /*params*/,
+    const std::map<std::string, std::string>& options,
+    const std::vector<std::string>& flags)
+    : LDBCommand(options, flags, false, BuildCmdLineOptions({ARG_JOB_DIR})) {
+  auto it = options.find(ARG_JOB_DIR);
+  if (it != options.end()) {
+    job_dir_ = it->second;
+  } else {
+    exec_state_ = LDBCommandExecuteResult::Failed("--job_dir is required");
+  }
+}
+
+void RemoteCompactionWorkerCommand::Help(std::string& ret) {
+  ret.append("  ");
+  ret.append(Name());
+  ret.append(" --db=<db_path> --job_dir=<dir>");
+  ret.append("\n    Read job_dir/input.bin, run DB::OpenAndCompact(), ");
+  ret.append("write job_dir/result.bin.\n");
+}
+
+void RemoteCompactionWorkerCommand::DoCommand() {
+  std::string input_path = job_dir_ + "/" + kInputFile;
+  std::string input;
+  if (!PollForFile(Env::Default(), input_path, &input)) {
+    exec_state_ = LDBCommandExecuteResult::Failed(
+        "Timed out (" + std::to_string(kPollTimeoutMs / 1000) +
+        "s) waiting for " + input_path);
+    return;
+  }
+
+  std::string output_dir = job_dir_ + "/output";
+  Status dir_s = Env::Default()->CreateDirIfMissing(output_dir);
+  if (!dir_s.ok()) {
+    exec_state_ = LDBCommandExecuteResult::Failed("CreateDirIfMissing: " +
+                                                  dir_s.ToString());
+    return;
+  }
+
+  std::string result;
+  CompactionServiceOptionsOverride override_options;
+  override_options.table_factory.reset(NewBlockBasedTableFactory());
+  Status s = DB::OpenAndCompact(OpenAndCompactOptions(), db_path_, output_dir,
+                                input, &result, override_options);
+  if (!s.ok()) {
+    exec_state_ =
+        LDBCommandExecuteResult::Failed("OpenAndCompact: " + s.ToString());
+    return;
+  }
+
+  std::string result_path = job_dir_ + "/" + kResultFile;
+  if (!AtomicWriteStringToFile(result_path, result)) {
+    exec_state_ =
+        LDBCommandExecuteResult::Failed("Failed to write " + result_path);
+    return;
+  }
+
+  fprintf(stdout, "remote_compaction_worker: OK (%zu bytes result)\n",
+          result.size());
+}
+
 }  // namespace ROCKSDB_NAMESPACE
@@ -835,4 +835,36 @@ class CompactionProgressDumpCommand : public LDBCommand {
  static const std::string ARG_PATH;
 };

+class RemoteCompactionPrimaryCommand : public LDBCommand {
+ public:
+  static std::string Name() { return "remote_compaction_primary"; }
+  RemoteCompactionPrimaryCommand(
+      const std::vector<std::string>& params,
+      const std::map<std::string, std::string>& options,
+      const std::vector<std::string>& flags);
+  void DoCommand() override;
+  bool NoDBOpen() override { return true; }
+  static void Help(std::string& ret);
+
+ private:
+  std::string job_dir_;
+  static const std::string ARG_JOB_DIR;
+};
+
+class RemoteCompactionWorkerCommand : public LDBCommand {
+ public:
+  static std::string Name() { return "remote_compaction_worker"; }
+  RemoteCompactionWorkerCommand(
+      const std::vector<std::string>& params,
+      const std::map<std::string, std::string>& options,
+      const std::vector<std::string>& flags);
+  void DoCommand() override;
+  bool NoDBOpen() override { return true; }
+  static void Help(std::string& ret);
+
+ private:
+  std::string job_dir_;
+  static const std::string ARG_JOB_DIR;
+};
+
 }  // namespace ROCKSDB_NAMESPACE
@@ -145,6 +145,8 @@ void LDBCommandRunner::PrintHelp(const LDBOptions& ldb_options,
  WriteExternalSstFilesCommand::Help(ret);
  IngestExternalSstFilesCommand::Help(ret);
  UnsafeRemoveSstFileCommand::Help(ret);
+  RemoteCompactionPrimaryCommand::Help(ret);
+  RemoteCompactionWorkerCommand::Help(ret);

  fprintf(to_stderr ? stderr : stdout, "%s\n", ret.c_str());
 }
@@ -0,0 +1 @@
+StringToMap (rocksdb/convenience.h) now preserves the outer braces of nested values so each map entry is in self-contained form and can be embedded directly into another `key=value;` string (e.g. SetOptions) without losing inner ';' delimiters. A new symmetric MapToString utility is provided.
@@ -0,0 +1 @@
+Fixed a bug where `AbortAllCompactions()` could leave automatic compaction work unscheduled when aborting before the background worker picked it, causing compaction and write stalls until DB restart.
@@ -0,0 +1 @@
+Fix a bug where db had a false positive compaction corruption error due to remote compaction service result with `has_accurate_num_input_records=false` not being serialized. 
@@ -0,0 +1 @@
+Reduce the likelihood of user-facing metadata operations blocking on MANIFEST rotation when file creation is slow, such as on remote storage. MANIFEST write batches containing foreground edits from `CreateColumnFamily()`, `DropColumnFamily()`, `CreateColumnFamilyWithImport()`, `IngestExternalFile()`, and `DeleteFilesInRanges()` now get a relaxed file size threshold for triggering MANIFEST rotation; background-only MANIFEST write batches, such as compaction and flush, continue to use the normal threshold. This change should make auto-tuning manifest file size more attractive (see `max_manifest_file_size` and `max_manifest_space_amp_pct` options).
@@ -521,7 +521,7 @@ class PresetCompressionDictTest
      public testing::WithParamInterface<std::tuple<CompressionType, bool>> {
 public:
  PresetCompressionDictTest()
-      : DBTestBase("db_test2", false /* env_do_fsync */),
+      : DBTestBase("compression_test_preset_dict", false /* env_do_fsync */),
        compression_type_(std::get<0>(GetParam())),
        bottommost_(std::get<1>(GetParam())) {}

@@ -44,6 +44,25 @@ struct IODispatcherImplData {
  virtual void ReleaseMemory(size_t bytes) = 0;
 };

+#ifndef NDEBUG
+static bool BlockIndicesAreSortedByOffset(
+    const std::vector<BlockHandle>& block_handles,
+    const std::vector<size_t>& block_indices) {
+  uint64_t prev_offset = 0;
+  for (size_t i = 0; i < block_indices.size(); ++i) {
+    if (block_indices[i] >= block_handles.size()) {
+      return false;
+    }
+    const uint64_t current_offset = block_handles[block_indices[i]].offset();
+    if (i > 0 && current_offset < prev_offset) {
+      return false;
+    }
+    prev_offset = current_offset;
+  }
+  return true;
+}
+#endif  // NDEBUG
+
 // Helper function to create and pin a block from a buffer
 // Used by both ReadSet::PollAndProcessAsyncIO and IODispatcherImpl::Impl
 static Status CreateAndPinBlockFromBuffer(
@@ -479,7 +498,8 @@ struct IODispatcherImpl::Impl : public IODispatcherImplData,
                        const std::vector<size_t>& block_indices);

  // Pre-coalesce blocks into groups, respecting max_group_bytes size limit.
-  // Returns groups ordered by first block index (earlier blocks first).
+  // block_indices must be sorted by block offset.
+  // Returns groups ordered by block offset.
  std::vector<CoalescedPrefetchGroup> PreCoalesceBlocks(
      const std::shared_ptr<IOJob>& job, const std::shared_ptr<ReadSet>& rs,
      const std::vector<size_t>& block_indices, size_t max_group_bytes);
@@ -697,16 +717,22 @@ Status IODispatcherImpl::Impl::SubmitJob(const std::shared_ptr<IOJob>& job,
  for (size_t i = 0; i < job->block_handles.size(); ++i) {
    rs->sorted_block_indices_[i] = i;
  }
-  std::sort(rs->sorted_block_indices_.begin(), rs->sorted_block_indices_.end(),
-            [&job](size_t a, size_t b) {
-              return job->block_handles[a].offset() <
-                     job->block_handles[b].offset();
-            });
+  if (!job->job_options.block_handles_are_sorted) {
+    TEST_SYNC_POINT("IODispatcherImpl::SubmitJob:SortBlockHandles");
+    std::sort(rs->sorted_block_indices_.begin(),
+              rs->sorted_block_indices_.end(), [&job](size_t a, size_t b) {
+                return job->block_handles[a].offset() <
+                       job->block_handles[b].offset();
+              });
+  }

-  // Step 1: Check cache and pin cached blocks
+  // Step 1: Check cache and pin cached blocks. Iterate in offset order so all
+  // downstream private helpers can assume block index vectors are already
+  // sorted.
  std::vector<size_t> block_indices_to_read;
+  block_indices_to_read.reserve(job->block_handles.size());

-  for (size_t i = 0; i < job->block_handles.size(); ++i) {
+  for (size_t i : rs->sorted_block_indices_) {
    const auto& data_block_handle = job->block_handles[i];

    // Lookup and pin block in cache
@@ -816,17 +842,11 @@ void IODispatcherImpl::Impl::PrepareIORequests(
    const std::vector<BlockHandle>& block_handles,
    std::vector<FSReadRequest>* read_reqs,
    std::vector<std::vector<size_t>>* coalesced_block_indices) {
-  // This is necessary because block handles may not be in sorted order
-  std::vector<size_t> sorted_block_indices = block_indices_to_read;
-  std::sort(sorted_block_indices.begin(), sorted_block_indices.end(),
-            [&block_handles](size_t a, size_t b) {
-              return block_handles[a].offset() < block_handles[b].offset();
-            });
-
+  assert(BlockIndicesAreSortedByOffset(block_handles, block_indices_to_read));
  assert(coalesced_block_indices->empty());
  coalesced_block_indices->resize(1);

-  for (const auto& block_idx : sorted_block_indices) {
+  for (const auto& block_idx : block_indices_to_read) {
    if (!coalesced_block_indices->back().empty()) {
      // Check if we can coalesce with previous block
      const auto& last_block_handle =
@@ -883,17 +903,12 @@ std::vector<CoalescedPrefetchGroup> IODispatcherImpl::Impl::PreCoalesceBlocks(
  const auto& block_handles = job->block_handles;
  const uint64_t coalesce_threshold = job->job_options.io_coalesce_threshold;

-  // Sort block indices by offset for coalescing
-  std::vector<size_t> sorted_indices = block_indices;
-  std::sort(sorted_indices.begin(), sorted_indices.end(),
-            [&block_handles](size_t a, size_t b) {
-              return block_handles[a].offset() < block_handles[b].offset();
-            });
+  assert(BlockIndicesAreSortedByOffset(block_handles, block_indices));

  // Build coalesced groups respecting max_group_bytes
  groups.emplace_back();

-  for (size_t idx : sorted_indices) {
+  for (size_t idx : block_indices) {
    size_t block_size = rs->block_sizes_[idx];

    // Skip blocks that are individually larger than the memory budget
@@ -10,6 +10,7 @@

 #include "rocksdb/io_dispatcher.h"

+#include <algorithm>
 #include <memory>
 #include <mutex>
 #include <thread>
@@ -836,6 +837,85 @@ TEST_F(IODispatcherTest, VerifyCoalescing) {
  }
 }

+TEST_F(IODispatcherTest, SortedBlockHandlesSkipDispatcherSorts) {
+  struct SortCounters {
+    int submit_job = 0;
+  };
+
+  auto install_callbacks = [](SortCounters* counters) {
+    SyncPoint::GetInstance()->SetCallBack(
+        "IODispatcherImpl::SubmitJob:SortBlockHandles",
+        [counters](void*) { ++counters->submit_job; });
+    SyncPoint::GetInstance()->EnableProcessing();
+  };
+
+  auto submit_job = [](IODispatcher* dispatcher, BlockBasedTable* table,
+                       const std::vector<BlockHandle>& block_handles,
+                       bool sorted) {
+    auto job = std::make_shared<IOJob>();
+    job->block_handles = block_handles;
+    job->table = table;
+    job->job_options.block_handles_are_sorted = sorted;
+    job->job_options.read_options.async_io = false;
+    job->job_options.io_coalesce_threshold = 1024 * 1024;
+
+    std::shared_ptr<ReadSet> read_set;
+    ASSERT_OK(dispatcher->SubmitJob(job, &read_set));
+    ASSERT_NE(read_set, nullptr);
+    for (size_t i = 0; i < block_handles.size(); ++i) {
+      CachableEntry<Block> block;
+      ASSERT_OK(read_set->ReadIndex(i, &block));
+      ASSERT_NE(block.GetValue(), nullptr);
+      ASSERT_GT(block.GetValue()->size(), 0);
+
+      InternalKeyComparator internal_comparator(BytewiseComparator());
+      std::unique_ptr<DataBlockIter> iter(block.GetValue()->NewDataIterator(
+          internal_comparator.user_comparator(), kDisableGlobalSequenceNumber));
+      iter->SeekToFirst();
+      ASSERT_TRUE(iter->Valid());
+
+      ParsedInternalKey parsed_key;
+      ASSERT_OK(
+          ParseInternalKey(iter->key(), &parsed_key, true /* log_err_key */));
+      ASSERT_TRUE(parsed_key.user_key.starts_with("key"));
+      ASSERT_OK(iter->status());
+    }
+  };
+
+  std::unique_ptr<BlockBasedTable> sorted_table;
+  std::vector<BlockHandle> sorted_handles;
+  ASSERT_OK(CreateAndOpenSST(10, &sorted_table, &sorted_handles));
+  ASSERT_GE(sorted_handles.size(), 5);
+  sorted_handles.resize(5);
+
+  SortCounters sorted_counters;
+  install_callbacks(&sorted_counters);
+  std::unique_ptr<IODispatcher> sorted_dispatcher(NewIODispatcher());
+  submit_job(sorted_dispatcher.get(), sorted_table.get(), sorted_handles,
+             true /* sorted */);
+  SyncPoint::GetInstance()->DisableProcessing();
+  SyncPoint::GetInstance()->ClearAllCallBacks();
+
+  EXPECT_EQ(sorted_counters.submit_job, 0);
+
+  std::unique_ptr<BlockBasedTable> unsorted_table;
+  std::vector<BlockHandle> unsorted_handles;
+  ASSERT_OK(CreateAndOpenSST(10, &unsorted_table, &unsorted_handles));
+  ASSERT_GE(unsorted_handles.size(), 5);
+  unsorted_handles.resize(5);
+  std::reverse(unsorted_handles.begin(), unsorted_handles.end());
+
+  SortCounters unsorted_counters;
+  install_callbacks(&unsorted_counters);
+  std::unique_ptr<IODispatcher> unsorted_dispatcher(NewIODispatcher());
+  submit_job(unsorted_dispatcher.get(), unsorted_table.get(), unsorted_handles,
+             false /* sorted */);
+  SyncPoint::GetInstance()->DisableProcessing();
+  SyncPoint::GetInstance()->ClearAllCallBacks();
+
+  EXPECT_EQ(unsorted_counters.submit_job, 1);
+}
+
 // Test that verifies the read request offsets and lengths match the
 // expected block handles.
 TEST_F(IODispatcherTest, VerifyReadRequestDetails) {
@@ -251,21 +251,22 @@ std::string UnescapeOptionString(const std::string& escaped_string) {
 }

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

 bool EndsWith(const std::string& string, const std::string& pattern) {
@@ -27,6 +27,31 @@ TEST(StringUtilTest, NumberToHumanString) {
  ASSERT_EQ("-10G", NumberToHumanString(-10000000000));
 }

+TEST(StringUtilTest, Trim) {
+  // Empty input.
+  EXPECT_EQ("", trim(""));
+  // No whitespace to strip.
+  EXPECT_EQ("a", trim("a"));
+  EXPECT_EQ("abc", trim("abc"));
+  // Leading whitespace only.
+  EXPECT_EQ("a", trim(" a"));
+  EXPECT_EQ("abc", trim("   abc"));
+  // Trailing whitespace only.
+  EXPECT_EQ("a", trim("a "));
+  EXPECT_EQ("abc", trim("abc   "));
+  // Both ends.
+  EXPECT_EQ("a", trim(" a "));
+  EXPECT_EQ("abc", trim("   abc   "));
+  EXPECT_EQ("a b c", trim("  a b c  "));  // interior whitespace preserved
+  // All-whitespace inputs of various lengths must trim to empty.
+  EXPECT_EQ("", trim(" "));
+  EXPECT_EQ("", trim("  "));
+  EXPECT_EQ("", trim("   "));
+  EXPECT_EQ("", trim("\t"));
+  EXPECT_EQ("", trim("\n"));
+  EXPECT_EQ("", trim(" \t\n\r"));
+}
+
 }  // namespace ROCKSDB_NAMESPACE

 int main(int argc, char** argv) {
Author	SHA1	Message	Date
Hui Xiao	62f291824a	Add remote compaction format compatibility test to check_format_compatible.sh	2026-06-03 08:48:11 -07:00
Peter Dillinger	3883a8d05e	Rename db_test2 -> db_etc2_test (#14218 ) Summary: When searching/grepping through unit tests, it's convenient to use test.cc suffix to match all unit tests, or to exclude test.cc when excluding unit tests from a code search. (I've even seen my AI assistant grep through `*test.cc`.) So I'm renaming db_test2.cc (as planned in https://github.com/facebook/rocksdb/issues/14076) Pull Request resolved: https://github.com/facebook/rocksdb/pull/14218 Test Plan: existing tests + CI Reviewed By: xingbowang Differential Revision: D90140624 Pulled By: pdillinger fbshipit-source-id: 78aa099290670bbb4093b2c7c02bc47ab3bf5f8e	2026-06-02 16:18:05 -07:00
Josh Kang	768de6e50d	Fix false-positive corruption error with remote compaction (#14808 ) Summary: Fix a false-positive compaction corruption error in the remote compaction path. Remote workers can mark `CompactionJobStats::num_input_records` as unreliable when input iteration uses seek/skip behavior, but the remote result serialization was dropping `has_accurate_num_input_records`. The primary then deserialized the flag as its default `true`, trusted an unreliable zero input-record count, and raised `Compaction number of input keys does not match number of keys processed`. This change serializes the accuracy flag with the rest of `CompactionJobStats` so the primary preserves the remote worker's "do not verify this count" signal. It also adds a targeted regression test and a release note for the bug fix. Pull Request resolved: https://github.com/facebook/rocksdb/pull/14808 Test Plan: New regression test that triggers corruption error without the fix by modifying `has_accurate_num_input_records=false` Reviewed By: xingbowang Differential Revision: D107128357 Pulled By: joshkang97 fbshipit-source-id: 3243c9c2ab18534652737f7410fe3c51724db549	2026-06-02 11:13:54 -07:00
Peter Dillinger	62f05627be	Reduce manifest rotation for foreground metadata ops (#14797 ) Summary: Async WAL precreation in https://github.com/facebook/rocksdb/pull/14738 / D105020559 was motivated by slow file creation time on remote storage. MANIFEST does not need the same precreation treatment as WAL because most MANIFEST writes come from background flush and compaction work, but user-facing metadata operations can still pay MANIFEST rotation file creation latency inline. File ingestion performance is a particular concern to some Meta users. Relax the effective MANIFEST rotation limit by 25% for MANIFEST write batches containing any foreground VersionEdit, while keeping background-only flush/compaction batches on the configured or auto-tuned limit. This covers column family manipulation, external file ingestion and import, and DeleteFilesInRange(s). SetOptions remains expected to avoid MANIFEST writes; the test keeps a regression guard for that behavior. The relaxation is intentionally bounded. It reduces the chance that foreground metadata operations create a new MANIFEST inline, while still allowing foreground operations to rotate once the current MANIFEST is beyond the relaxed threshold. Heavier blocking operations like manual Flush or CompactRange already trigger additional file creation and do not get this treatment here, though that could be reconsidered later. This should reduce a potential latency hazard of manifest file size auto-tuning: more frequent MANIFEST rotations. With this change, rotation latency is shifted toward background-only MANIFEST batches when possible. Pull Request resolved: https://github.com/facebook/rocksdb/pull/14797 Test Plan: Expanded DBEtc3Test.AutoTuneManifestSize to cover the foreground threshold behavior and the original auto-tuning behavior in separate phases: - verifies foreground-only CreateColumnFamily writes get only bounded 25% headroom by asserting the first four large-CF additions do not rotate and the fifth does; - verifies auto-tuned background thresholds still prevent excessive rotation; - verifies foreground operations stay below the relaxed threshold for CreateColumnFamily, IngestExternalFile, CreateColumnFamilyWithImport, and DeleteFilesInRanges; - verifies SetOptions still does not write to MANIFEST; - verifies a following background flush still rotates at the normal threshold; - preserves the persisted compacted manifest size close/reopen coverage. Reviewed By: xingbowang Differential Revision: D106578771 Pulled By: pdillinger fbshipit-source-id: f8e274032cd9e7f50e95b685c949242f95351498	2026-06-01 18:13:29 -07:00
Peter Dillinger	9ef369c606	Make StringToMap entries self-contained for direct round-trip (#14805 ) Summary: StringToMap previously stripped the outer braces from nested values, making single-entry round-trip via `key=value;` (e.g. SetOptions) silently corrupt values that contain ';'. For example, a filter_policy with sub-options serialized as filter_policy={id=ribbonfilter:10:-1;bloom_before_level=-1;} parsed to map[filter_policy] = "id=ribbonfilter:10:-1;bloom_before_level=-1;", and embedding that map entry directly into another `key=value;` string re-exposed the inner ';' as a top-level delimiter. This change is a step toward the "essential view": the outer `{...}` of a nested value is part of the value's identity. StringToMap preserves it, so each map entry is in self-contained form (a simple value or a single balanced `{...}` block) and can be embedded directly without further escaping by the caller. However some permissiveness is kept for compatibility: list-style typed parsers (ParseVector, ParseArray, kStringMap, the listener parser) and scalar dispatch in OptionTypeInfo::Parse accept either the bare or the wrapped form via a new OptionTypeInfo::StripOuterBraces helper that peels at most one outer-pair-matched layer. So braced scalar input like `key={42};` and `key={true};` still parses, and `key={};` denotes an empty list, distinct from `key={{}};` which denotes a one-element list with an empty element. A new public MapToString is added to convenience.h as the symmetric inverse of StringToMap: a trivial `key=value;` joiner that relies on each value already being self-contained (which StringToMap guarantees). Also fixed and refactored our trim() function because it would do the wrong thing for a single space. This problem was detected by expanding the StringToMapRandomTest based on code review feedback, and should only improve existing callers to trim(). The OPTIONS-file serializer code is untouched, so the on-disk format is byte-for-byte identical and format-compatibility is preserved. Bonus: fixes / clarifications to CLAUDE.md's build-system note. Pull Request resolved: https://github.com/facebook/rocksdb/pull/14805 Test Plan: db_bloom_filter_test extends MutableFilterPolicy with the OPTIONS-file format for filter_policy to confirm that form round-trips through SetOptions. options_test updates StringToMapTest expectations for the new brace-preserving behavior, and adds: - MapToStringTest covering the new joiner. - StringToMapMapToStringRoundTripTest including a single-entry pull-and-embed scenario that previously corrupted on round-trip. - FullOptionsStringToMapRoundTripTest that drives a populated DBOptions and CFOptions through GetStringFromXxx -> StringToMap -> per-entry single round-trip -> MapToString -> GetXxxFromString -> VerifyXxxOptions, catching any custom serializer that emits a value with ';' without enclosing it in '{}'. - EmptyBracedVectorAndBracedScalarTest covering `key={};` (empty list) and braced scalar input like `key={42};`, `key={true};`. - Expanded StringToMapRandomTest with round-tripping. Some explicit trim() tests added to string_util_test.cc Format compatibility verified with SHORT_TEST=1 tools/check_format_compatible.sh Other existing tests in options_test, customizable_test, options_settable_test, listener_test, options_file_test, options_util_test, db_options_test, and compaction_service_test have extensive coverage of serialization / deserialization logic. Reviewed By: hx235, xingbowang Differential Revision: D106855466 Pulled By: pdillinger fbshipit-source-id: 872aac8d819c7b90c807b92bab85569b48fa2aa0	2026-06-01 17:07:11 -07:00
Xingbo Wang	30dba7f41a	Fix compaction abort rescheduling before queue pick (#14800 ) Summary: - AbortAllCompactions can race with an already-scheduled automatic compaction before the background worker calls PickCompactionFromQueue(). MaybeScheduleFlushOrCompaction() has already consumed one unscheduled_compactions_ credit when it scheduled the worker, but the CF is still present in compaction_queue_ with queued_for_compaction=true. If the worker returns kCompactionAborted before popping the CF, ResumeAllCompactions() can see unscheduled_compactions_ == 0 and fail to schedule the still-queued work, leaving compaction permanently stalled until DB restart. - Restore the unscheduled compaction credit in the non-prepicked abort path, matching the existing BG-work-stopped handling for the same scheduled-before-pick state. Prepicked/manual compactions are not adjusted because they do not represent an unpopped automatic compaction_queue_ entry whose scheduling credit was consumed. - Add AbortScheduledAutomaticCompactionBeforePick to deterministically reproduce the lost-credit race with a sync point at BackgroundCallCompaction:0. The test verifies that ResumeAllCompactions() schedules the still-queued automatic compaction by checking COMPACT_WRITE_BYTES advances and L0 file count drops. Pull Request resolved: https://github.com/facebook/rocksdb/pull/14800 Test Plan: - Without the implementation fix, `DBCompactionAbortTest.AbortScheduledAutomaticCompactionBeforePick` fails because `COMPACT_WRITE_BYTES` remains unchanged after `ResumeAllCompactions()`. Reviewed By: pdillinger Differential Revision: D106684155 Pulled By: xingbowang fbshipit-source-id: 6dacea473ef481eb3122eb15e296e5b69635413f	2026-06-01 16:11:22 -07:00
Xingbo Wang	023fbb074a	Optimize MultiScan dispatch for sorted blocks (#14783 ) Summary: - Propagate validated, sorted MultiScan range state from `DBIter::Prepare()` through `MultiScanArgs`. - Mark block-based table IO jobs as already sorted when the public MultiScan ranges have been validated. - Keep `IODispatcher::SubmitJob()` as the normalization boundary for unsorted callers, while allowing sorted callers to skip the defensive block-handle sort. - Update private dispatcher coalescing helpers to consume sorted block indices and add debug assertions for that precondition. ## Testing CI, new unit test Pull Request resolved: https://github.com/facebook/rocksdb/pull/14783 Reviewed By: anand1976 Differential Revision: D106301516 Pulled By: xingbowang fbshipit-source-id: 99b7ffcaecbf27cb79f15feb4af8680ff1e422d9	2026-06-01 15:36:38 -07:00
				`@@ -0,0 +1 @@`
				StringToMap (rocksdb/convenience.h) now preserves the outer braces of nested values so each map entry is in self-contained form and can be embedded directly into another `key=value;` string (e.g. SetOptions) without losing inner ';' delimiters. A new symmetric MapToString utility is provided.
				`@@ -0,0 +1 @@`
				Fixed a bug where `AbortAllCompactions()` could leave automatic compaction work unscheduled when aborting before the background worker picked it, causing compaction and write stalls until DB restart.
				`@@ -0,0 +1 @@`
				Fix a bug where db had a false positive compaction corruption error due to remote compaction service result with `has_accurate_num_input_records=false` not being serialized.
				`@@ -0,0 +1 @@`
				Reduce the likelihood of user-facing metadata operations blocking on MANIFEST rotation when file creation is slow, such as on remote storage. MANIFEST write batches containing foreground edits from `CreateColumnFamily()`, `DropColumnFamily()`, `CreateColumnFamilyWithImport()`, `IngestExternalFile()`, and `DeleteFilesInRanges()` now get a relaxed file size threshold for triggering MANIFEST rotation; background-only MANIFEST write batches, such as compaction and flush, continue to use the normal threshold. This change should make auto-tuning manifest file size more attractive (see `max_manifest_file_size` and `max_manifest_space_amp_pct` options).