Add option to verify block checksums of output files (facebook#14103)

jaykorean · jaykorean · commit 4de3ce3b0160 · 2025-11-10T15:38:32.000-08:00
Summary: For all compactions, RocksDB performs a lightweight sanity check on output SST files before installation (in `CompactionJob::VerifyOutputFiles()`). However, this lightweight check may not catch corruption that is small enough to allow the SST files to still be opened. There is an existing feature, `paranoid_file_check`, which opens the SST file, iterates through all keys, and checks the hash of each key. While this provides the ultimate level of data integrity checking, it comes at a high computational cost. In this PR, we introduce a new mutable CF option, `verify_output_flags`. The `verify_output_flags` is a bitmask enum that allows users to select various verification types, including block checksum verification, full key iteration, and file checksum verification (to be added in subsequent PRs). Note that the existing `paranoid_file_check` option is equivalent to a full key iteration check. Block-level checksum verification is much lighter than the full key iteration check. Please note that the previously deprecated `verify_checksums_in_compaction` option (removed in version 5.3.0) was for verifying the checksum of **input SST files**. RocksDB continues to perform this verification for both local and remote compactions, and this behavior remains unchanged. In contrast, this PR focuses on verifying the **output SST files**. ## To follow up - File-level Checksum verification for output SST files - Deprecate `paranoid_file_checks` option in favor of the new option - Add to stress test / db_bench Pull Request resolved: facebook#14103 Test Plan: New Unit Test added. The corruption is both detected by `paranoid_file_check` and various types of verification set by this new option, `verify_output_flags` ``` ./compaction_service_test --gtest_filter="*CompactionServiceTest.CorruptedOutput*" ``` Reviewed By: pdillinger Differential Revision: D86357924 Pulled By: jaykorean fbshipit-source-id: a9e04798f249c7e977231e179622a0830d6675fe
diff --git a/db/compaction/compaction_job.cc b/db/compaction/compaction_job.cc
@@ -812,20 +812,19 @@ Status CompactionJob::SyncOutputDirectories() {
 Status CompactionJob::VerifyOutputFiles() {
   Status status;
   std::vector<port::Thread> thread_pool;
-  std::vector<const CompactionOutputs::Output*> files_output;
-  for (const auto& state : compact_->sub_compact_states) {
-    for (const auto& output : state.GetOutputs()) {
-      files_output.emplace_back(&output);
-    }
-  }
   ColumnFamilyData* cfd = compact_->compaction->column_family_data();
-  std::atomic<size_t> next_file_idx(0);
-  auto verify_table = [&](Status& output_status) {
-    while (true) {
-      size_t file_idx = next_file_idx.fetch_add(1);
-      if (file_idx >= files_output.size()) {
-        break;
-      }
+  VerifyOutputFlags verify_output_flags =
+      compact_->compaction->mutable_cf_options().verify_output_flags;
+
+  // For backward compatibility
+  if (paranoid_file_checks_) {
+    verify_output_flags |= VerifyOutputFlags::kVerifyIteration;
+    verify_output_flags |= VerifyOutputFlags::kEnableForLocalCompaction;
+    verify_output_flags |= VerifyOutputFlags::kEnableForRemoteCompaction;
+  }
+
+  auto verify_table = [&](SubcompactionState& subcompaction_state) {
+    for (const auto& output_file : subcompaction_state.GetOutputs()) {
       // Verify that the table is usable
       // We set for_compaction to false and don't
       // OptimizeForCompactionTableRead here because this is a special case
@@ -834,13 +833,19 @@ Status CompactionJob::VerifyOutputFiles() {
       // verification as user reads since the goal is to cache it here for
       // further user reads
       ReadOptions verify_table_read_options(Env::IOActivity::kCompaction);
+      verify_table_read_options.verify_checksums = true;
+      verify_table_read_options.readahead_size =
+          file_options_for_read_.compaction_readahead_size;
+
+      std::unique_ptr<TableReader> table_reader_guard;
+      TableReader* table_reader_ptr = table_reader_guard.get();
       verify_table_read_options.rate_limiter_priority =
           GetRateLimiterPriority();
       InternalIterator* iter = cfd->table_cache()->NewIterator(
           verify_table_read_options, file_options_, cfd->internal_comparator(),
-          files_output[file_idx]->meta,
+          output_file.meta,
           /*range_del_agg=*/nullptr, compact_->compaction->mutable_cf_options(),
-          /*table_reader_ptr=*/nullptr,
+          /*table_reader_ptr=*/&table_reader_ptr,
           cfd->internal_stats()->GetFileReadHist(
               compact_->compaction->output_level()),
           TableReaderCaller::kCompactionRefill, /*arena=*/nullptr,
@@ -850,38 +855,63 @@ Status CompactionJob::VerifyOutputFiles() {
           /*largest_compaction_key=*/nullptr,
           /*allow_unprepared_value=*/false);
       auto s = iter->status();
-
-      if (s.ok() && paranoid_file_checks_) {
-        OutputValidator validator(cfd->internal_comparator(),
-                                  /*_enable_hash=*/true);
-        for (iter->SeekToFirst(); iter->Valid(); iter->Next()) {
-          s = validator.Add(iter->key(), iter->value());
-          if (!s.ok()) {
-            break;
+      if (s.ok()) {
+        // Check for remote/local compaction and verify_output_flags flags
+        const bool should_verify =
+            (subcompaction_state.compaction_job_stats.is_remote_compaction &&
+             !!(verify_output_flags &
+                VerifyOutputFlags::kEnableForRemoteCompaction)) ||
+            (!subcompaction_state.compaction_job_stats.is_remote_compaction &&
+             !!(verify_output_flags &
+                VerifyOutputFlags::kEnableForLocalCompaction));
+
+        if (should_verify) {
+          const bool should_verify_block_checksum =
+              !!(verify_output_flags & VerifyOutputFlags::kVerifyBlockChecksum);
+          const bool should_verify_iteration =
+              !!(verify_output_flags & VerifyOutputFlags::kVerifyIteration);
+          if (should_verify_block_checksum) {
+            assert(table_reader_ptr != nullptr);
+            // If verifying iteration as well, verify meta blocks here only to
+            // avoid redundant checks on data blocks
+            s = table_reader_ptr->VerifyChecksum(
+                verify_table_read_options, TableReaderCaller::kCompaction,
+                /*meta_blocks_only=*/should_verify_iteration);
+          }
+          if (s.ok() && should_verify_iteration) {
+            OutputValidator validator(cfd->internal_comparator(),
+                                      /*_enable_hash=*/true);
+            for (iter->SeekToFirst(); iter->Valid(); iter->Next()) {
+              s = validator.Add(iter->key(), iter->value());
+              if (!s.ok()) {
+                break;
+              }
+            }
+            if (s.ok()) {
+              s = iter->status();
+            }
+            if (s.ok() && !validator.CompareValidator(output_file.validator)) {
+              s = Status::Corruption(
+                  "Key-value checksum of compaction output doesn't match what "
+                  "was computed when written");
+            }
           }
-        }
-        if (s.ok()) {
-          s = iter->status();
-        }
-        if (s.ok() &&
-            !validator.CompareValidator(files_output[file_idx]->validator)) {
-          s = Status::Corruption("Paranoid checksums do not match");
         }
       }
 
       delete iter;
 
       if (!s.ok()) {
-        output_status = s;
+        subcompaction_state.status = s;
         break;
       }
     }
   };
   for (size_t i = 1; i < compact_->sub_compact_states.size(); i++) {
     thread_pool.emplace_back(verify_table,
-                             std::ref(compact_->sub_compact_states[i].status));
+                             std::ref(compact_->sub_compact_states[i]));
   }
-  verify_table(compact_->sub_compact_states[0].status);
+  verify_table(compact_->sub_compact_states[0]);
   for (auto& thread : thread_pool) {
     thread.join();
   }
diff --git a/db/compaction/compaction_service_test.cc b/db/compaction/compaction_service_test.cc
@@ -1047,6 +1047,7 @@ TEST_F(CompactionServiceTest, CorruptedOutputParanoidFileCheck) {
     Destroy(options);
     options.disable_auto_compactions = true;
     options.paranoid_file_checks = paranoid_file_check_enabled;
+    options.verify_output_flags = VerifyOutputFlags::kVerifyNone;
     ReopenWithCompactionService(&options);
     GenerateTestData();
 
@@ -1101,6 +1102,87 @@ TEST_F(CompactionServiceTest, CorruptedOutputParanoidFileCheck) {
   }
 }
 
+TEST_F(CompactionServiceTest, CorruptedOutputVerifyOutputFlags) {
+  for (VerifyOutputFlags verify_output_flags :
+       {VerifyOutputFlags::kVerifyNone,
+        VerifyOutputFlags::kEnableForLocalCompaction |
+            VerifyOutputFlags::kVerifyBlockChecksum,
+        VerifyOutputFlags::kEnableForRemoteCompaction |
+            VerifyOutputFlags::kVerifyBlockChecksum,
+        VerifyOutputFlags::kEnableForRemoteCompaction |
+            VerifyOutputFlags::kVerifyIteration,
+        VerifyOutputFlags::kVerifyAll}) {
+    SCOPED_TRACE(
+        "verify_output_flags=" +
+        std::to_string(static_cast<std::underlying_type_t<VerifyOutputFlags>>(
+            verify_output_flags)));
+
+    Options options = CurrentOptions();
+    Destroy(options);
+    options.disable_auto_compactions = true;
+    options.paranoid_file_checks = false;
+    options.verify_output_flags = verify_output_flags;
+    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->output_files.empty());
+          // Corrupt files here
+          for (const auto& output_file : compaction_result->output_files) {
+            std::string file_name =
+                compaction_result->output_path + "/" + output_file.file_name;
+
+            // Corrupt very small range of bytes. This corruption is so small
+            // that this isn't caught by default light-weight check
+            ASSERT_OK(test::CorruptFile(env_, file_name, 0, 1,
+                                        false /* verifyChecksum */));
+          }
+        });
+    SyncPoint::GetInstance()->EnableProcessing();
+    const bool is_enabled_for_remote_compaction =
+        !!(verify_output_flags & VerifyOutputFlags::kEnableForRemoteCompaction);
+    const bool should_verify_block_checksum =
+        !!(verify_output_flags & VerifyOutputFlags::kVerifyBlockChecksum);
+    const bool should_verify_iteration =
+        !!(verify_output_flags & VerifyOutputFlags::kVerifyIteration);
+
+    Status s = db_->CompactRange(CompactRangeOptions(), &start, &end);
+    if (is_enabled_for_remote_compaction &&
+        (should_verify_block_checksum || should_verify_iteration)) {
+      ASSERT_NOK(s);
+      ASSERT_TRUE(s.IsCorruption());
+    } else {
+      // CompactRange() goes through if block checksum wasn't verified
+      ASSERT_OK(s);
+    }
+
+    ASSERT_GE(my_cs->GetCompactionNum(), comp_num + 1);
+
+    SyncPoint::GetInstance()->DisableProcessing();
+    SyncPoint::GetInstance()->ClearAllCallBacks();
+
+    // On the worker side, the compaction is considered success
+    // Verification is done on the primary side
+    CompactionServiceResult result;
+    my_cs->GetResult(&result);
+    ASSERT_OK(result.status);
+    ASSERT_TRUE(result.stats.is_manual_compaction);
+    ASSERT_TRUE(result.stats.is_remote_compaction);
+  }
+}
+
 TEST_F(CompactionServiceTest, TruncatedOutput) {
   Options options = CurrentOptions();
   options.disable_auto_compactions = true;
diff --git a/include/rocksdb/advanced_options.h b/include/rocksdb/advanced_options.h
@@ -156,6 +156,61 @@ enum class PrepopulateBlobCache : uint8_t {
   kFlushOnly = 0x1,  // Prepopulate blobs during flush only
 };
 
+// Bitmask enum for verify output flags during compaction.
+// This allows fine-grained control over what verification is performed
+// on compaction output files and when it's enabled.
+enum class VerifyOutputFlags : uint32_t {
+  kVerifyNone = 0x0,  // No verification
+
+  // First set of bits: type of verifications
+  kVerifyBlockChecksum = 1 << 0,  // Verify block checksums
+  kVerifyIteration = 1 << 1,      // Verify iteration and full key/value hash
+                                  // by comparing the one inserted into a
+                                  // file, and what is read back.
+
+  // TODO - Implement
+  // kVerifyFileChecksum = 1 << 2,   // Verify file-level checksum
+
+  // Second set of bits: when to enable verification
+  kEnableForLocalCompaction = 1 << 10,   // Enable for local compaction
+  kEnableForRemoteCompaction = 1 << 11,  // Enable for remote compaction
+
+  // TODO - Implement
+  // kEnableForFlush = 1 << 12,  // Enable for flush
+
+  kVerifyAll = 0xFFFFFFFF,
+};
+
+inline VerifyOutputFlags operator|(VerifyOutputFlags lhs,
+                                   VerifyOutputFlags rhs) {
+  using T = std::underlying_type_t<VerifyOutputFlags>;
+  return static_cast<VerifyOutputFlags>(static_cast<T>(lhs) |
+                                        static_cast<T>(rhs));
+}
+
+inline VerifyOutputFlags& operator|=(VerifyOutputFlags& lhs,
+                                     VerifyOutputFlags rhs) {
+  lhs = lhs | rhs;
+  return lhs;
+}
+
+inline VerifyOutputFlags operator&(VerifyOutputFlags lhs,
+                                   VerifyOutputFlags rhs) {
+  using T = std::underlying_type_t<VerifyOutputFlags>;
+  return static_cast<VerifyOutputFlags>(static_cast<T>(lhs) &
+                                        static_cast<T>(rhs));
+}
+
+inline VerifyOutputFlags& operator&=(VerifyOutputFlags& lhs,
+                                     VerifyOutputFlags rhs) {
+  lhs = lhs & rhs;
+  return lhs;
+}
+
+inline bool operator!(VerifyOutputFlags flag) {
+  return flag == VerifyOutputFlags::kVerifyNone;
+}
+
 struct AdvancedColumnFamilyOptions {
   // The maximum number of write buffers that are built up in memory.
   // The default and the minimum number is 2, so that when 1 write buffer
@@ -704,6 +759,13 @@ struct AdvancedColumnFamilyOptions {
   // Dynamically changeable through SetOptions() API
   bool paranoid_file_checks = false;
 
+  // Bitmask enum for output verification option.
+  //
+  // Default: 0 (kVerifyNone)
+  //
+  // Dynamically changeable (as a uint32_t) through SetOptions() API.
+  VerifyOutputFlags verify_output_flags = VerifyOutputFlags::kVerifyNone;
+
   // In debug mode, RocksDB runs consistency checks on the LSM every time the
   // LSM changes (Flush, Compaction, AddFile). When this option is true, these
   // checks are also enabled in release mode. These checks were historically
diff --git a/options/cf_options.cc b/options/cf_options.cc
@@ -395,6 +395,10 @@ static std::unordered_map<std::string, OptionTypeInfo>
          {offsetof(struct MutableCFOptions, paranoid_file_checks),
           OptionType::kBoolean, OptionVerificationType::kNormal,
           OptionTypeFlags::kMutable}},
+        {"verify_output_flags",
+         {offsetof(struct MutableCFOptions, verify_output_flags),
+          OptionType::kUInt32T, OptionVerificationType::kNormal,
+          OptionTypeFlags::kMutable}},
         {"verify_checksums_in_compaction",
          {0, OptionType::kBoolean, OptionVerificationType::kDeprecated,
           OptionTypeFlags::kMutable}},
diff --git a/options/cf_options.h b/options/cf_options.h
@@ -143,6 +143,7 @@ struct MutableCFOptions {
         preclude_last_level_data_seconds(
             options.preclude_last_level_data_seconds),
         preserve_internal_time_seconds(options.preserve_internal_time_seconds),
+        verify_output_flags(options.verify_output_flags),
         enable_blob_files(options.enable_blob_files),
         min_blob_size(options.min_blob_size),
         blob_file_size(options.blob_file_size),
@@ -213,6 +214,7 @@ struct MutableCFOptions {
         compaction_options_fifo(),
         preclude_last_level_data_seconds(0),
         preserve_internal_time_seconds(0),
+        verify_output_flags(VerifyOutputFlags::kVerifyNone),
         enable_blob_files(false),
         min_blob_size(0),
         blob_file_size(0),
@@ -313,6 +315,7 @@ struct MutableCFOptions {
   CompactionOptionsUniversal compaction_options_universal;
   uint64_t preclude_last_level_data_seconds;
   uint64_t preserve_internal_time_seconds;
+  VerifyOutputFlags verify_output_flags;
 
   // Blob file related options
   bool enable_blob_files;
diff --git a/options/options_helper.cc b/options/options_helper.cc
@@ -270,6 +270,8 @@ void UpdateColumnFamilyOptions(const MutableCFOptions& moptions,
   cf_opts->compaction_options_fifo = moptions.compaction_options_fifo;
   cf_opts->compaction_options_universal = moptions.compaction_options_universal;
 
+  cf_opts->verify_output_flags = moptions.verify_output_flags;
+
   // Blob file related options
   cf_opts->enable_blob_files = moptions.enable_blob_files;
   cf_opts->min_blob_size = moptions.min_blob_size;
diff --git a/options/options_settable_test.cc b/options/options_settable_test.cc
@@ -687,7 +687,8 @@ TEST_F(OptionsSettableTest, ColumnFamilyOptionsAllFieldsSettable) {
       "memtable_veirfy_per_key_checksum_on_seek=1;"
       "memtable_op_scan_flush_trigger=123;"
       "memtable_avg_op_scan_flush_trigger=12;"
-      "cf_allow_ingest_behind=1;",
+      "cf_allow_ingest_behind=1;"
+      "verify_output_flags=2049;",
       new_options));
 
   ASSERT_NE(new_options->blob_cache.get(), nullptr);
diff --git a/options/options_test.cc b/options/options_test.cc
@@ -1720,13 +1720,24 @@ TEST_F(OptionsTest, MutableCFOptions) {
 
   ASSERT_OK(GetColumnFamilyOptionsFromString(
       config_options, cf_opts,
-      "paranoid_file_checks=true; block_based_table_factory.block_align=false; "
+      "paranoid_file_checks=true; "
+      "verify_output_flags=2049; "
+      "block_based_table_factory.block_align=false; "
       "block_based_table_factory.super_block_alignment_size=65536; "
       "block_based_table_factory.super_block_alignment_space_overhead_ratio="
       "4096; "
       "block_based_table_factory.block_size=8192;",
       &cf_opts));
   ASSERT_TRUE(cf_opts.paranoid_file_checks);
+  ASSERT_NE(
+      (cf_opts.verify_output_flags & VerifyOutputFlags::kVerifyBlockChecksum),
+      VerifyOutputFlags::kVerifyNone);
+  ASSERT_NE((cf_opts.verify_output_flags &
+             VerifyOutputFlags::kEnableForRemoteCompaction),
+            VerifyOutputFlags::kVerifyNone);
+  ASSERT_EQ((cf_opts.verify_output_flags &
+             VerifyOutputFlags::kEnableForLocalCompaction),
+            VerifyOutputFlags::kVerifyNone);
   ASSERT_NE(cf_opts.table_factory.get(), nullptr);
   auto* bbto = cf_opts.table_factory->GetOptions<BlockBasedTableOptions>();
   ASSERT_NE(bbto, nullptr);
@@ -2581,6 +2592,7 @@ TEST_F(OptionsOldApiTest, GetOptionsFromMapTest) {
   ASSERT_EQ(new_cf_opt.optimize_filters_for_hits, true);
   ASSERT_EQ(new_cf_opt.prefix_extractor->AsString(), "rocksdb.FixedPrefix.31");
   ASSERT_EQ(new_cf_opt.experimental_mempurge_threshold, 0.003);
+  ASSERT_EQ(new_cf_opt.verify_output_flags, VerifyOutputFlags::kVerifyNone);
   ASSERT_EQ(new_cf_opt.enable_blob_files, true);
   ASSERT_EQ(new_cf_opt.min_blob_size, 1ULL << 10);
   ASSERT_EQ(new_cf_opt.blob_file_size, 1ULL << 30);
diff --git a/table/block_based/block_based_table_reader.cc b/table/block_based/block_based_table_reader.cc
diff --git a/table/block_based/block_based_table_reader.h b/table/block_based/block_based_table_reader.h
diff --git a/table/external_table.cc b/table/external_table.cc
diff --git a/table/table_reader.h b/table/table_reader.h
