Skip building remote compaction output file when not OK (#13183)

Summary:
During `FinishCompactionOutputFile()` if there's an IOError, we may end up having the output in memory, but table properties are not populated, because `outputs.UpdateTableProperties();` is called only when `s.ok()` is true.

However, during remote compaction result serialization, we always try to access the `table_properties` which may be null.  This was causing a segfault.

We can skip building the output files in the result completely if the status is not ok.

# Unit Test
New test added
```
./compaction_service_test --gtest_filter="*CompactionOutputFileIOError*"
```

Before the fix
```
Received signal 11 (Segmentation fault)
Invoking GDB for stack trace...
#4  0x00000000004708ed in rocksdb::TableProperties::TableProperties (this=0x7fae070fb4e8) at ./include/rocksdb/table_properties.h:212
212     struct TableProperties {
#5  0x00007fae0b195b9e in rocksdb::CompactionServiceOutputFile::CompactionServiceOutputFile (this=0x7fae070fb400, name=..., smallest=0, largest=0, _smallest_internal_key=..., _largest_internal_key=..., _oldest_ancester_time=1733335023, _file_creation_time=1733335026, _epoch_number=1, _file_checksum=..., _file_checksum_func_name=..., _paranoid_hash=0, _marked_for_compaction=false, _unique_id=..., _table_properties=...) at ./db/compaction/compaction_job.h:450
450             table_properties(_table_properties) {}
```

After the fix
```
[ RUN      ] CompactionServiceTest.CompactionOutputFileIOError
[       OK ] CompactionServiceTest.CompactionOutputFileIOError (4499 ms)
[----------] 1 test from CompactionServiceTest (4499 ms total)

[----------] Global test environment tear-down
[==========] 1 test from 1 test case ran. (4499 ms total)
[  PASSED  ] 1 test.
```

Pull Request resolved: #13183

Reviewed By: anand1976

Differential Revision: D66770876

Pulled By: jaykorean

fbshipit-source-id: 63df7c2786ce0353f38a93e493ae4e7b591f4ed9

Author: jaykorean

db/compaction/compaction_job.cc

@@ -1586,6 +1586,8 @@ Status CompactionJob::FinishCompactionOutputFile(
   const uint64_t current_entries = outputs.NumEntries();
 
   s = outputs.Finish(s, seqno_to_time_mapping_);
+  TEST_SYNC_POINT_CALLBACK(
+      "CompactionJob::FinishCompactionOutputFile()::AfterFinish", &s);
 
   if (s.ok()) {
     // With accurate smallest and largest key, we can get a slightly more

db/compaction/compaction_service_job.cc

@@ -376,16 +376,18 @@ Status CompactionServiceCompactionJob::Run() {
   // Build Output
   compaction_result_->output_level = compact_->compaction->output_level();
   compaction_result_->output_path = output_path_;
-  for (const auto& output_file : sub_compact->GetOutputs()) {
-    auto& meta = output_file.meta;
-    compaction_result_->output_files.emplace_back(
-        MakeTableFileName(meta.fd.GetNumber()), meta.fd.smallest_seqno,
-        meta.fd.largest_seqno, meta.smallest.Encode().ToString(),
-        meta.largest.Encode().ToString(), meta.oldest_ancester_time,
-        meta.file_creation_time, meta.epoch_number, meta.file_checksum,
-        meta.file_checksum_func_name, output_file.validator.GetHash(),
-        meta.marked_for_compaction, meta.unique_id,
-        *output_file.table_properties);
+  if (status.ok()) {
+    for (const auto& output_file : sub_compact->GetOutputs()) {
+      auto& meta = output_file.meta;
+      compaction_result_->output_files.emplace_back(
+          MakeTableFileName(meta.fd.GetNumber()), meta.fd.smallest_seqno,
+          meta.fd.largest_seqno, meta.smallest.Encode().ToString(),
+          meta.largest.Encode().ToString(), meta.oldest_ancester_time,
+          meta.file_creation_time, meta.epoch_number, meta.file_checksum,
+          meta.file_checksum_func_name, output_file.validator.GetHash(),
+          meta.marked_for_compaction, meta.unique_id,
+          *output_file.table_properties);
+    }
   }
 
   TEST_SYNC_POINT_CALLBACK("CompactionServiceCompactionJob::Run:0",

db/compaction/compaction_service_test.cc

@@ -416,6 +416,38 @@ TEST_F(CompactionServiceTest, ManualCompaction) {
   ASSERT_TRUE(result.stats.is_remote_compaction);
 }
 
+TEST_F(CompactionServiceTest, CompactionOutputFileIOError) {
+  Options options = CurrentOptions();
+  options.disable_auto_compactions = true;
+  ReopenWithCompactionService(&options);
+  GenerateTestData();
+
+  auto my_cs = GetCompactionService();
+
+  SyncPoint::GetInstance()->SetCallBack(
+      "CompactionJob::FinishCompactionOutputFile()::AfterFinish",
+      [&](void* status) {
+        // override status
+        auto s = static_cast<Status*>(status);
+        *s = Status::IOError("Injected IOError!");
+      });
+  SyncPoint::GetInstance()->EnableProcessing();
+
+  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();
+  ASSERT_NOK(db_->CompactRange(CompactRangeOptions(), &start, &end));
+  ASSERT_GE(my_cs->GetCompactionNum(), comp_num + 1);
+
+  CompactionServiceResult result;
+  my_cs->GetResult(&result);
+  ASSERT_NOK(result.status);
+  ASSERT_TRUE(result.stats.is_manual_compaction);
+  ASSERT_TRUE(result.stats.is_remote_compaction);
+}
+
 TEST_F(CompactionServiceTest, PreservedOptionsLocalCompaction) {
   Options options = CurrentOptions();
   options.level0_file_num_compaction_trigger = 2;