Add remote compaction format compatibility test to check_format_compatible.sh

tools/check_format_compatible.sh

@@ -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
+# This way, we only do at most n+1 checkout+build steps.  The one
-# need to stash away executables.
+# 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

tools/generate_random_db.sh

@@ -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
+# A shell script to load some pre generated data file to a DB using ldb tool.
-# ./ldb needs to be avaible to be executed.
+# <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

tools/ldb_cmd.cc

@@ -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

tools/ldb_cmd_impl.h

@@ -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

tools/ldb_tool.cc

@@ -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());
 }