Skip Wal Recovery on SecondaryDB Open if for Remote Compaction (#14462)

Summary:
Skip WAL recovery when opening a secondary DB instance in OpenAndCompact() for remote compaction. WAL replay is unnecessary in this flow since only LSM state from MANIFEST is needed.

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

Test Plan:
- make -j db_secondary_test && ./db_secondary_test — 35/35 passed
- make -j compaction_service_test && ./compaction_service_test — 43/43 passed (includes new SkipWALRecoveryInOpenAndCompact test)
- make -j options_settable_test && ./options_settable_test --gtest_filter="*DBOptionsAllFieldsSettable*" — 1/1 passed
- Removed temporary hack in stress test that disables WAL

Reviewed By: hx235

Differential Revision: D96788211

Pulled By: jaykorean

fbshipit-source-id: f91a2f861f2450ebc83423ed4c6f5b70da7d9e8b

db/compaction/compaction_service_test.cc

@@ -409,6 +409,34 @@ TEST_F(CompactionServiceTest, BasicCompactions) {
   SyncPoint::GetInstance()->DisableProcessing();
 }
 
+TEST_F(CompactionServiceTest, SkipWALRecoveryInOpenAndCompact) {
+  // Verify that OpenAndCompact skips WAL recovery when opening the secondary
+  // instance. WAL replay is unnecessary for remote compaction since it only
+  // needs the LSM state from MANIFEST.
+  Options options = CurrentOptions();
+  ReopenWithCompactionService(&options);
+
+  // Track whether FindAndRecoverLogFiles is called during compaction
+  std::atomic_bool wal_recovery_called{false};
+  SyncPoint::GetInstance()->SetCallBack(
+      "DBImplSecondary::FindAndRecoverLogFiles:Begin",
+      [&](void* /* arg */) { wal_recovery_called.store(true); });
+  SyncPoint::GetInstance()->EnableProcessing();
+
+  // Generate data and trigger compaction (which uses OpenAndCompact)
+  GenerateTestData();
+  ASSERT_OK(dbfull()->TEST_WaitForCompact());
+
+  // WAL recovery should NOT have been called during OpenAndCompact
+  ASSERT_FALSE(wal_recovery_called.load());
+
+  // Data should still be correct (compaction worked without WAL recovery)
+  VerifyTestData();
+
+  SyncPoint::GetInstance()->DisableProcessing();
+  SyncPoint::GetInstance()->ClearAllCallBacks();
+}
+
 TEST_F(CompactionServiceTest, ManualCompaction) {
   Options options = CurrentOptions();
   options.disable_auto_compactions = true;

db/db_impl/db_impl.h

@@ -1744,6 +1744,7 @@ class DBImpl : public DB {
 
  private:
   friend class DB;
+  friend class DBImplSecondary;
   friend class ErrorHandler;
   friend class InternalStats;
   friend class PessimisticTransaction;

db/db_impl/db_impl_secondary.cc

@@ -43,7 +43,6 @@ Status DBImplSecondary::Recover(
     RecoveryContext* /*recovery_ctx*/, bool* /*can_retry*/) {
   mutex_.AssertHeld();
 
-  JobContext job_context(0);
   Status s;
   s = static_cast<ReactiveVersionSet*>(versions_.get())
           ->Recover(column_families, &manifest_reader_, &manifest_reporter_,
@@ -61,24 +60,12 @@ Status DBImplSecondary::Recover(
     max_total_in_memory_state_ += mutable_cf_options.write_buffer_size *
                                   mutable_cf_options.max_write_buffer_number;
   }
-  if (s.ok()) {
-    default_cf_handle_ = new ColumnFamilyHandleImpl(
-        versions_->GetColumnFamilySet()->GetDefault(), this, &mutex_);
-    default_cf_internal_stats_ = default_cf_handle_->cfd()->internal_stats();
+  default_cf_handle_ = new ColumnFamilyHandleImpl(
+      versions_->GetColumnFamilySet()->GetDefault(), this, &mutex_);
+  default_cf_internal_stats_ = default_cf_handle_->cfd()->internal_stats();
 
-    std::unordered_set<ColumnFamilyData*> cfds_changed;
-    s = FindAndRecoverLogFiles(&cfds_changed, &job_context);
-  }
-
-  if (s.IsPathNotFound()) {
-    ROCKS_LOG_INFO(immutable_db_options_.info_log,
-                   "Secondary tries to read WAL, but WAL file(s) have already "
-                   "been purged by primary.");
-    s = Status::OK();
-  }
   // TODO: update options_file_number_ needed?
 
-  job_context.Clean();
   return s;
 }
 
@@ -88,6 +75,7 @@ Status DBImplSecondary::FindAndRecoverLogFiles(
     JobContext* job_context) {
   assert(nullptr != cfds_changed);
   assert(nullptr != job_context);
+  TEST_SYNC_POINT("DBImplSecondary::FindAndRecoverLogFiles:Begin");
   Status s;
   std::vector<uint64_t> logs;
   s = FindNewLogNumbers(&logs);
@@ -740,6 +728,17 @@ Status DB::OpenAsSecondary(
     const std::string& secondary_path,
     const std::vector<ColumnFamilyDescriptor>& column_families,
     std::vector<ColumnFamilyHandle*>* handles, std::unique_ptr<DB>* dbptr) {
+  return DBImplSecondary::OpenAsSecondaryImpl(
+      db_options, dbname, secondary_path, column_families, handles, dbptr,
+      /*recover_wal=*/true);
+}
+
+Status DBImplSecondary::OpenAsSecondaryImpl(
+    const DBOptions& db_options, const std::string& dbname,
+    const std::string& secondary_path,
+    const std::vector<ColumnFamilyDescriptor>& column_families,
+    std::vector<ColumnFamilyHandle*>* handles, std::unique_ptr<DB>* dbptr,
+    bool recover_wal) {
   *dbptr = nullptr;
 
   DBOptions tmp_opts(db_options);
@@ -784,7 +783,21 @@ Status DB::OpenAsSecondary(
   impl->wal_in_db_path_ = impl->immutable_db_options_.IsWalDirSameAsDBPath();
 
   impl->mutex_.Lock();
+  JobContext job_context(0);
   s = impl->Recover(column_families, true, false, false);
+  // WAL recovery is optional: DB::OpenAsSecondary() needs it to replay
+  // memtable data, while DB::OpenAndCompact() skips it since remote
+  // compaction only needs LSM state from MANIFEST.
+  if (s.ok() && recover_wal) {
+    std::unordered_set<ColumnFamilyData*> cfds_changed;
+    s = impl->FindAndRecoverLogFiles(&cfds_changed, &job_context);
+    if (s.IsPathNotFound()) {
+      ROCKS_LOG_INFO(impl->immutable_db_options_.info_log,
+                     "Secondary tries to read WAL, but WAL file(s) have "
+                     "already been purged by primary.");
+      s = Status::OK();
+    }
+  }
   if (s.ok()) {
     for (const auto& cf : column_families) {
       auto cfd =
@@ -806,6 +819,7 @@ Status DB::OpenAsSecondary(
   }
   impl->mutex_.Unlock();
   sv_context.Clean();
+  job_context.Clean();
   if (s.ok()) {
     dbptr->reset(impl);
     for (auto h : *handles) {
@@ -1539,11 +1553,13 @@ Status DB::OpenAndCompact(
     }
   }
 
-  // 5. Open db As Secondary
+  // 5. Open db As Secondary (skip WAL recovery — remote compaction only
+  //    needs LSM state from MANIFEST, not memtable data from WAL replay)
   std::unique_ptr<DB> db;
   std::vector<ColumnFamilyHandle*> handles;
-  s = DB::OpenAsSecondary(db_options, name, output_directory, column_families,
-                          &handles, &db);
+  s = DBImplSecondary::OpenAsSecondaryImpl(db_options, name, output_directory,
+                                           column_families, &handles, &db,
+                                           /*recover_wal=*/false);
   if (!s.ok()) {
     return s;
   }

db/db_impl/db_impl_secondary.h

@@ -78,8 +78,9 @@ class DBImplSecondary : public DBImpl {
                   std::string secondary_path);
   ~DBImplSecondary() override;
 
-  // Recover by replaying MANIFEST and WAL. Also initialize manifest_reader_
-  // and log_readers_ to facilitate future operations.
+  // Recover by replaying MANIFEST only. Also initialize manifest_reader_
+  // to facilitate future operations. WAL recovery, if needed, is done
+  // separately after opening (see DB::OpenAsSecondary).
   Status Recover(const std::vector<ColumnFamilyDescriptor>& column_families,
                  bool read_only, bool error_if_wal_file_exists,
                  bool error_if_data_exists_in_wals, bool is_retry = false,
@@ -384,6 +385,17 @@ class DBImplSecondary : public DBImpl {
   uint64_t CalculateResumedCompactionBytes(
       const CompactionProgress& compaction_progress) const;
 
+  // Internal helper for opening a secondary instance. Recover() replays
+  // MANIFEST only. When recover_wal is true, WAL files are also replayed
+  // (needed by DB::OpenAsSecondary). When false, WAL replay is skipped
+  // (used by DB::OpenAndCompact which only needs LSM state).
+  static Status OpenAsSecondaryImpl(
+      const DBOptions& db_options, const std::string& dbname,
+      const std::string& secondary_path,
+      const std::vector<ColumnFamilyDescriptor>& column_families,
+      std::vector<ColumnFamilyHandle*>* handles, std::unique_ptr<DB>* dbptr,
+      bool recover_wal);
+
   // Cache log readers for each log number, used for continue WAL replay
   // after recovery
   std::map<uint64_t, std::unique_ptr<LogReaderContainer>> log_readers_;

db_stress_tool/db_stress_test_base.cc

@@ -3708,13 +3708,6 @@ void StressTest::Open(SharedState* shared, bool reopen) {
 
   // Remote Compaction
   if (FLAGS_remote_compaction_worker_threads > 0) {
-    // TODO(jaykorean) Remove this after fix - remote worker shouldn't recover
-    // from WAL
-    if (!FLAGS_disable_wal) {
-      fprintf(stderr,
-              "WAL is not compatible with Remote Compaction in Stress Test\n");
-      exit(1);
-    }
     if ((options_.enable_blob_files ||
          options_.enable_blob_garbage_collection ||
          FLAGS_allow_setting_blob_options_dynamically)) {

tools/db_crashtest.py

@@ -434,9 +434,7 @@ default_params = {
     # (block checksum, iteration, file checksum), bits 10-11 are when to
     # enable (local compaction, remote compaction). 0x407 = all types +
     # local, 0xC07 = all types + local + remote, 0xFFFFFFFF = all.
-    "verify_output_flags": lambda: random.choice(
-        [0] * 3 + [0x407, 0xC07, 0xFFFFFFFF]
-    ),
+    "verify_output_flags": lambda: random.choice([0] * 3 + [0x407, 0xC07, 0xFFFFFFFF]),
     "paranoid_memory_checks": lambda: random.choice([0] * 7 + [1]),
     "memtable_veirfy_per_key_checksum_on_seek": lambda: random.choice([0] * 7 + [1]),
     "memtable_batch_lookup_optimization": lambda: random.randint(0, 1),
@@ -878,8 +876,6 @@ def finalize_and_sanitize(src_params):
         dest_params["enable_blob_files"] = 0
         dest_params["enable_blob_garbage_collection"] = 0
         dest_params["allow_setting_blob_options_dynamically"] = 0
-        # TODO Fix - Remote worker shouldn't recover from WAL
-        dest_params["disable_wal"] = 1
         # Disable Incompatible Ones
         dest_params["inplace_update_support"] = 0
         dest_params["checkpoint_one_in"] = 0
@@ -901,8 +897,6 @@ def finalize_and_sanitize(src_params):
         dest_params["open_write_fault_one_in"] = 0
         dest_params["open_read_fault_one_in"] = 0
         dest_params["sync_fault_injection"] = 0
-    else:
-        dest_params["allow_resumption_one_in"] = 0
 
     # UDI now supports all operation types (Put, Delete, Merge, etc.).
     # Only parallel compression and mmap_read remain incompatible.
@@ -1185,9 +1179,6 @@ def finalize_and_sanitize(src_params):
             # have to disable metadata write fault injection to other file
             dest_params["exclude_wal_from_write_fault_injection"] = 1
             dest_params["metadata_write_fault_one_in"] = 0
-
-            # TODO Fix - Remote worker shouldn't recover from WAL
-            dest_params["remote_compaction_worker_threads"] = 0
     # Disabling block align if mixed manager is being used
     if dest_params.get("compression_manager") == "custom":
         if dest_params.get("block_align") == 1:
@@ -1292,6 +1283,10 @@ def finalize_and_sanitize(src_params):
     if dest_params["inplace_update_support"] == 1:
         dest_params["memtable_veirfy_per_key_checksum_on_seek"] = 0
 
+    # allow_resumption requires remote compaction
+    if dest_params.get("remote_compaction_worker_threads", 0) == 0:
+        dest_params["allow_resumption_one_in"] = 0
+
     return dest_params
 
 
@@ -1407,7 +1402,13 @@ def execute_cmd(cmd, timeout=None, timeout_pstack=False):
             print("KILLED %d (SIGTERM did not work)\n" % child.pid)
             outs, errs = child.communicate()
 
-    return hit_timeout, child.returncode, outs.decode("utf-8"), errs.decode("utf-8"), pid
+    return (
+        hit_timeout,
+        child.returncode,
+        outs.decode("utf-8"),
+        errs.decode("utf-8"),
+        pid,
+    )
 
 
 def print_output_and_exit_on_error(stdout, stderr, print_stderr_separately=False):

unreleased_history/behavior_changes/skip_wal_recovery_remote_compaction

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