Fix false positive corruption in compaction output verification (facebook#14456)

Summary:
Fix a bug where `VerifyOutputFiles()` produces false positive "Key-value checksum of compaction output doesn't match what was computed when written" errors when `verify_output_flags` includes `kVerifyIteration` but `paranoid_file_checks` is false.

The root cause is a hash enable flag mismatch between writing and verification:

- During compaction writing (`OpenCompactionOutputFile`), the `OutputValidator` hash computation was gated solely by `paranoid_file_checks_`. When false, `enable_hash=false` and the hash stays at 0.
- During verification (`VerifyOutputFiles`), a new `OutputValidator` is always created with `enable_hash=true`, computing a non-zero hash.
- `CompareValidator()` then compares 0 vs non-zero, producing a false positive corruption.

This was exposed by the crash test randomization of `verify_output_flags` added in facebook#14433. Before that change, `verify_output_flags` was always 0, so `kVerifyIteration` was only exercised via `paranoid_file_checks=true` (which correctly enabled the hash during writing).

The fix ensures hash computation is enabled during writing whenever either `paranoid_file_checks_` is true OR `verify_output_flags` includes `kVerifyIteration`.

Pull Request resolved: facebook#14456

Test Plan:
- Added `DBCompactionTest.VerifyIterationWithoutParanoidFileChecks`
- Added `DBCompactionTest.VerifyAllOutputFlagsWithoutParanoidFileChecks`
- Round-trip verified: tests FAIL without fix, PASS with fix

Reviewed By: jaykorean

Differential Revision: D96371769

Pulled By: xingbowang

fbshipit-source-id: 2f7406327496c7b541e4fa2668894df89eb813e8

Author: xingbowang

db/compaction/compaction_job.cc

@@ -2494,8 +2494,16 @@ Status CompactionJob::OpenCompactionOutputFile(SubcompactionState* sub_compact,
       return s;
     }
 
+    // Enable hash computation if paranoid_file_checks is on or if
+    // verify_output_flags includes kVerifyIteration, so that
+    // VerifyOutputFiles() can compare the hash of the written data
+    // against a re-read of the output file.
+    bool enable_output_hash =
+        paranoid_file_checks_ ||
+        !!(sub_compact->compaction->mutable_cf_options().verify_output_flags &
+           VerifyOutputFlags::kVerifyIteration);
     outputs.AddOutput(std::move(meta), cfd->internal_comparator(),
-                      paranoid_file_checks_);
+                      enable_output_hash);
   }
 
   writable_file->SetIOPriority(GetRateLimiterPriority());
@@ -2928,12 +2936,17 @@ void CompactionJob::RestoreCompactionOutputs(
 
   const auto& output_files = subcompaction_progress_per_level.GetOutputFiles();
 
+  const bool enable_output_hash =
+      paranoid_file_checks_ ||
+      !!(compact_->compaction->mutable_cf_options().verify_output_flags &
+         VerifyOutputFlags::kVerifyIteration);
+
   for (size_t i = 0; i < output_files.size(); i++) {
     FileMetaData file_copy = output_files[i];
 
     outputs_to_restore->AddOutput(std::move(file_copy),
                                   cfd->internal_comparator(),
-                                  paranoid_file_checks_, true /* finished */);
+                                  enable_output_hash, true /* finished */);
 
     outputs_to_restore->UpdateTableProperties(
         *output_files_table_properties[i]);

db/db_compaction_test.cc

@@ -11790,6 +11790,82 @@ TEST_F(DBCompactionTest, VerifyFileChecksumOnCompactionOutput) {
   SyncPoint::GetInstance()->ClearAllCallBacks();
 }
 
+// Regression test: verify_output_flags with kVerifyIteration should work
+// correctly even when paranoid_file_checks is false. Before the fix, the
+// OutputValidator hash was only computed during writing when
+// paranoid_file_checks was true, but the verification always computed the
+// hash, leading to a false positive "Key-value checksum of compaction output
+// doesn't match" error.
+TEST_F(DBCompactionTest, VerifyIterationWithoutParanoidFileChecks) {
+  Options options = CurrentOptions();
+  options.disable_auto_compactions = true;
+  options.paranoid_file_checks = false;
+  options.verify_output_flags = VerifyOutputFlags::kVerifyIteration |
+                                VerifyOutputFlags::kEnableForLocalCompaction;
+  DestroyAndReopen(options);
+
+  // Create 2 L0 files to trigger compaction
+  for (int i = 0; i < 10; i++) {
+    ASSERT_OK(Put(Key(i), "value" + std::to_string(i)));
+  }
+  ASSERT_OK(Flush());
+
+  for (int i = 5; i < 15; i++) {
+    ASSERT_OK(Put(Key(i), "value2_" + std::to_string(i)));
+  }
+  ASSERT_OK(Flush());
+
+  // Compaction should succeed without false corruption errors
+  ASSERT_OK(db_->CompactRange(CompactRangeOptions(), nullptr, nullptr));
+
+  // Verify data is intact
+  for (int i = 0; i < 15; i++) {
+    std::string expected;
+    if (i >= 5) {
+      expected = "value2_" + std::to_string(i);
+    } else {
+      expected = "value" + std::to_string(i);
+    }
+    ASSERT_EQ(Get(Key(i)), expected);
+  }
+}
+
+// Also test all verification types combined without paranoid_file_checks
+TEST_F(DBCompactionTest, VerifyAllOutputFlagsWithoutParanoidFileChecks) {
+  Options options = CurrentOptions();
+  options.disable_auto_compactions = true;
+  options.paranoid_file_checks = false;
+  options.file_checksum_gen_factory = GetFileChecksumGenCrc32cFactory();
+  options.verify_output_flags = VerifyOutputFlags::kVerifyBlockChecksum |
+                                VerifyOutputFlags::kVerifyIteration |
+                                VerifyOutputFlags::kVerifyFileChecksum |
+                                VerifyOutputFlags::kEnableForLocalCompaction;
+  DestroyAndReopen(options);
+
+  for (int i = 0; i < 10; i++) {
+    ASSERT_OK(Put(Key(i), "value" + std::to_string(i)));
+  }
+  ASSERT_OK(Flush());
+
+  for (int i = 5; i < 15; i++) {
+    ASSERT_OK(Put(Key(i), "value2_" + std::to_string(i)));
+  }
+  ASSERT_OK(Flush());
+
+  // Compaction should succeed with all verification types enabled
+  ASSERT_OK(db_->CompactRange(CompactRangeOptions(), nullptr, nullptr));
+
+  for (int i = 0; i < 15; i++) {
+    std::string expected;
+    if (i >= 5) {
+      expected = "value2_" + std::to_string(i);
+    } else {
+      expected = "value" + std::to_string(i);
+    }
+    ASSERT_EQ(Get(Key(i)), expected);
+  }
+}
+
 }  // namespace ROCKSDB_NAMESPACE
 
 int main(int argc, char** argv) {