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...
https://github.com/facebook/rocksdb/issues/4  0x00000000004708ed in rocksdb::TableProperties::TableProperties (this=0x7fae070fb4e8) at ./include/rocksdb/table_properties.h:212
212     struct TableProperties {
https://github.com/facebook/rocksdb/issues/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: https://github.com/facebook/rocksdb/pull/13183

Reviewed By: anand1976

Differential Revision: D66770876

Pulled By: jaykorean

fbshipit-source-id: 63df7c2786ce0353f38a93e493ae4e7b591f4ed9

HISTORY.md

@@ -1,6 +1,41 @@
 # Rocksdb Change Log
 > NOTE: Entries for next release do not go here. Follow instructions in `unreleased_history/README.txt`
 
+## 9.9.3 (12/03/2024)
+### Performance Improvements
+* In buffered IO mode, try to align writes on power of 2 if checksum handoff is not enabled for the file type being written.
+
+## 9.9.2 (11/22/2024)
+### Bug Fixes
+* Honor ConfigOptions.ignore_unknown_options in ParseStruct()
+
+## 9.9.1 (11/30/2024)
+### Behavior Changes
+* Updates the hidden hook RocksDbThreadYieldAndCheckAbort() to support MySQL to abort long-running query.
+
+## 9.9.0 (11/18/2024)
+### New Features
+* Multi-Column-Family-Iterator (CoalescingIterator/AttributeGroupIterator) is no longer marked as experimental
+* Adds a new table property "rocksdb.newest.key.time" which records the unix timestamp of the newest key. Uses this table property for FIFO TTL and temperature change compaction.
+
+### Public API Changes
+* Added a new API `Transaction::GetAttributeGroupIterator` that can be used to create a multi-column-family attribute group iterator over the specified column families, including the data from both the transaction and the underlying database. This API is currently supported for optimistic and write-committed pessimistic transactions.
+* Added a new API `Transaction::GetCoalescingIterator` that can be used to create a multi-column-family coalescing iterator over the specified column families, including the data from both the transaction and the underlying database. This API is currently supported for optimistic and write-committed pessimistic transactions.
+
+### Behavior Changes
+* `BaseDeltaIterator` now honors the read option `allow_unprepared_value`.
+
+### Bug Fixes
+* `BaseDeltaIterator` now calls `PrepareValue` on the base iterator in case it has been created with the `allow_unprepared_value` read option set. Earlier, such base iterators could lead to incorrect values being exposed from `BaseDeltaIterator`.
+* Fix a leak of obsolete blob files left open until DB::Close(). This bug was introduced in version 9.4.0.
+* Fix missing cases of corruption retry during DB open and read API processing.
+* Fix a bug for transaction db with 2pc where an old WAL may be retained longer than needed (#13127).
+* Fix leaks of some open SST files (until `DB::Close()`) that are written but never become live due to various failures. (We now have a check for such leaks with no outstanding issues.)
+* Fix a bug for replaying WALs for WriteCommitted transaction DB when its user-defined timestamps setting is toggled on/off between DB sessions.
+
+### Performance Improvements
+* Fix regression in issue #12038 due to `Options::compaction_readahead_size` greater than `max_sectors_kb` (i.e, largest I/O size that the OS issues to a block device defined in linux)
+
 ## 9.8.0 (10/25/2024)
 ### New Features
 * All non-`block_cache` options in `BlockBasedTableOptions` are now mutable with `DB::SetOptions()`. See also Bug Fixes below.

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;

db/db_impl/db_impl.cc

@@ -3247,8 +3247,13 @@ Status DBImpl::MultiGetImpl(
       s = Status::Aborted();
       break;
     }
+
     // This could be a long-running operation
-    ROCKSDB_THREAD_YIELD_HOOK();
+    bool aborted = ROCKSDB_THREAD_YIELD_CHECK_ABORT();
+    if (aborted) {
+      s = Status::Aborted("Query abort.");
+      break;
+    }
   }
 
   // Post processing (decrement reference counts and record statistics)

db/db_iter.cc

@@ -581,8 +581,14 @@ bool DBIter::FindNextUserEntryInternal(bool skipping_saved_key,
     } else {
       iter_.Next();
     }
+
     // This could be a long-running operation due to tombstones, etc.
-    ROCKSDB_THREAD_YIELD_HOOK();
+    bool aborted = ROCKSDB_THREAD_YIELD_CHECK_ABORT();
+    if (aborted) {
+      valid_ = false;
+      status_ = Status::Aborted("Query abort.");
+      return false;
+    }
   } while (iter_.Valid());
 
   valid_ = false;

db/external_sst_file_basic_test.cc

@@ -271,6 +271,79 @@ TEST_F(ExternalSSTFileBasicTest, Basic) {
   DestroyAndRecreateExternalSSTFilesDir();
 }
 
+TEST_F(ExternalSSTFileBasicTest, AlignedBufferedWrite) {
+  class AlignedWriteFS : public FileSystemWrapper {
+   public:
+    explicit AlignedWriteFS(const std::shared_ptr<FileSystem>& _target)
+        : FileSystemWrapper(_target) {}
+    ~AlignedWriteFS() override {}
+    const char* Name() const override { return "AlignedWriteFS"; }
+
+    IOStatus NewWritableFile(const std::string& fname, const FileOptions& opts,
+                             std::unique_ptr<FSWritableFile>* result,
+                             IODebugContext* dbg) override {
+      class AlignedWritableFile : public FSWritableFileOwnerWrapper {
+       public:
+        AlignedWritableFile(std::unique_ptr<FSWritableFile>& file)
+            : FSWritableFileOwnerWrapper(std::move(file)), last_write_(false) {}
+
+        using FSWritableFileOwnerWrapper::Append;
+        IOStatus Append(const Slice& data, const IOOptions& options,
+                        IODebugContext* dbg) override {
+          EXPECT_FALSE(last_write_);
+          if ((data.size() & (data.size() - 1)) != 0) {
+            last_write_ = true;
+          }
+          return target()->Append(data, options, dbg);
+        }
+
+       private:
+        bool last_write_;
+      };
+
+      std::unique_ptr<FSWritableFile> file;
+      IOStatus s = target()->NewWritableFile(fname, opts, &file, dbg);
+      if (s.ok()) {
+        result->reset(new AlignedWritableFile(file));
+      }
+      return s;
+    }
+  };
+
+  Options options = CurrentOptions();
+  std::shared_ptr<AlignedWriteFS> aligned_fs =
+      std::make_shared<AlignedWriteFS>(env_->GetFileSystem());
+  std::unique_ptr<Env> wrap_env(
+      new CompositeEnvWrapper(options.env, aligned_fs));
+  options.env = wrap_env.get();
+
+  EnvOptions env_options;
+  env_options.writable_file_max_buffer_size = 64 * 1024 * 1024;
+
+  SstFileWriter sst_file_writer(env_options, options);
+
+  // Current file size should be 0 after sst_file_writer init and before open a
+  // file.
+  ASSERT_EQ(sst_file_writer.FileSize(), 0);
+
+  // file1.sst (0 => 99)
+  std::string file1 = sst_files_dir_ + "file1.sst";
+  ASSERT_OK(sst_file_writer.Open(file1));
+  Random r(301);
+  for (int k = 0; k < 16 * 1024; k++) {
+    uint32_t num = 4096 + r.Uniform(8192);
+    std::string random_string = r.RandomString(num);
+    ASSERT_OK(sst_file_writer.Put(Key(k), random_string));
+  }
+  Status s = sst_file_writer.Finish();
+  ASSERT_OK(s) << s.ToString();
+
+  // Current file size should be non-zero after success write.
+  ASSERT_GT(sst_file_writer.FileSize(), 0);
+
+  DestroyAndRecreateExternalSSTFilesDir();
+}
+
 class ChecksumVerifyHelper {
  private:
   Options options_;

file/writable_file_writer.cc

@@ -107,6 +107,14 @@ IOStatus WritableFileWriter::Append(const IOOptions& opts, const Slice& data,
   // Flush only when buffered I/O
   if (!use_direct_io() && (buf_.Capacity() - buf_.CurrentSize()) < left) {
     if (buf_.CurrentSize() > 0) {
+      if (!buffered_data_with_checksum_) {
+        // If we're not calculating checksum of buffered data, fill the
+        // buffer before flushing so that the writes are aligned. This will
+        // benefit file system performance.
+        size_t appended = buf_.Append(src, left);
+        left -= appended;
+        src += appended;
+      }
       s = Flush(io_options);
       if (!s.ok()) {
         set_seen_error(s);

include/rocksdb/version.h

@@ -13,7 +13,7 @@
 // minor or major version number planned for release.
 #define ROCKSDB_MAJOR 9
 #define ROCKSDB_MINOR 9
-#define ROCKSDB_PATCH 0
+#define ROCKSDB_PATCH 3
 
 // Do not use these. We made the mistake of declaring macros starting with
 // double underscore. Now we have to live with our choice. We'll deprecate these

options/options_helper.cc

@@ -995,7 +995,8 @@ Status OptionTypeInfo::ParseStruct(
     std::unordered_map<std::string, std::string> unused;
     status =
         ParseType(config_options, opt_value, *struct_map, opt_addr, &unused);
-    if (status.ok() && !unused.empty()) {
+    if (status.ok() && !unused.empty() &&
+        !config_options.ignore_unknown_options) {
       status = Status::InvalidArgument(
           "Unrecognized option", struct_name + "." + unused.begin()->first);
     }
@@ -1006,7 +1007,7 @@ Status OptionTypeInfo::ParseStruct(
         Find(opt_name.substr(struct_name.size() + 1), *struct_map, &elem_name);
     if (opt_info != nullptr) {
       status = opt_info->Parse(config_options, elem_name, opt_value, opt_addr);
-    } else {
+    } else if (!config_options.ignore_unknown_options) {
       status = Status::InvalidArgument("Unrecognized option", opt_name);
     }
   } else {
@@ -1015,7 +1016,7 @@ Status OptionTypeInfo::ParseStruct(
     const auto opt_info = Find(opt_name, *struct_map, &elem_name);
     if (opt_info != nullptr) {
       status = opt_info->Parse(config_options, elem_name, opt_value, opt_addr);
-    } else {
+    } else if (!config_options.ignore_unknown_options) {
       status = Status::InvalidArgument("Unrecognized option",
                                        struct_name + "." + opt_name);
     }

options/options_test.cc

@@ -832,6 +832,21 @@ TEST_F(OptionsTest, CompressionOptionsFromString) {
   ASSERT_OK(GetColumnFamilyOptionsFromString(ignore, ColumnFamilyOptions(),
                                              "compression_opts.unknown=bad",
                                              &base_cf_opt));
+
+  // Test with some additional values added
+  config_options.ignore_unknown_options = false;
+  ASSERT_NOK(
+      GetColumnFamilyOptionsFromString(config_options, ColumnFamilyOptions(),
+                                       "new_unknown_field=whatever; "
+                                       "bottommost_compression_opts=4:5:6:7",
+                                       &base_cf_opt));
+  // Test Ignoring Unknown Options
+  config_options.ignore_unknown_options = true;
+  ASSERT_OK(
+      GetColumnFamilyOptionsFromString(config_options, ColumnFamilyOptions(),
+                                       "new_unknown_field=whatever; "
+                                       "bottommost_compression_opts=4:5:6:7",
+                                       &base_cf_opt));
 }
 
 TEST_F(OptionsTest, OldInterfaceTest) {

port/port.h

@@ -24,14 +24,10 @@
 // A temporary hook into long-running RocksDB threads to support modifying their
 // priority etc. This should become a public API hook once the requirements
 // are better understood.
-extern "C" void RocksDbThreadYield() __attribute__((__weak__));
-#define ROCKSDB_THREAD_YIELD_HOOK() \
-  {                                 \
-    if (RocksDbThreadYield) {       \
-      RocksDbThreadYield();         \
-    }                               \
-  }
+// Returns true if query is aborted.
+extern "C" bool RocksDbThreadYieldAndCheckAbort() __attribute__((__weak__));
+#define ROCKSDB_THREAD_YIELD_CHECK_ABORT() \
+  (RocksDbThreadYieldAndCheckAbort ? RocksDbThreadYieldAndCheckAbort() : false)
 #else
-#define ROCKSDB_THREAD_YIELD_HOOK() \
-  {}
+#define ROCKSDB_THREAD_YIELD_CHECK_ABORT() (false)
 #endif

unreleased_history/behavior_changes/base_delta_iterator_allow_unprepared_value.md

@@ -1 +0,0 @@
-`BaseDeltaIterator` now honors the read option `allow_unprepared_value`.

unreleased_history/bug_fixes/base_delta_iterator_prepare_value.md

@@ -1 +0,0 @@
-`BaseDeltaIterator` now calls `PrepareValue` on the base iterator in case it has been created with the `allow_unprepared_value` read option set. Earlier, such base iterators could lead to incorrect values being exposed from `BaseDeltaIterator`.

unreleased_history/bug_fixes/blob_file_leak.md

@@ -1 +0,0 @@
-* Fix a leak of obsolete blob files left open until DB::Close(). This bug was introduced in version 9.4.0.

unreleased_history/bug_fixes/missing_strong_reads.md

@@ -1 +0,0 @@
-Fix missing cases of corruption retry during DB open and read API processing.

unreleased_history/bug_fixes/old_wal_2pc.md

@@ -1 +0,0 @@
-* Fix a bug for transaction db with 2pc where an old WAL may be retained longer than needed (#13127). 

unreleased_history/bug_fixes/open_sst_file_leaks.md

@@ -1 +0,0 @@
-* Fix leaks of some open SST files (until `DB::Close()`) that are written but never become live due to various failures. (We now have a check for such leaks with no outstanding issues.)

unreleased_history/bug_fixes/write_committed_toggle_udt.md

@@ -1 +0,0 @@
-*Fix a bug for replaying WALs for WriteCommitted transaction DB when its user-defined timestamps setting is toggled on/off between DB sessions.

unreleased_history/new_features/multi_cf_iterator.md

@@ -1 +0,0 @@
-Multi-Column-Family-Iterator (CoalescingIterator/AttributeGroupIterator) is no longer marked as experimental

unreleased_history/new_features/tp_newest_key_time.md

@@ -1 +0,0 @@
-* Adds a new table property "rocksdb.newest.key.time" which records the unix timestamp of the newest key. Uses this table property for FIFO TTL and temperature change compaction. 

unreleased_history/performance_improvements/cra_regression.md

@@ -1 +0,0 @@
-Fix regression in issue #12038 due to `Options::compaction_readahead_size` greater than `max_sectors_kb` (i.e, largest I/O size that the OS issues to a block device defined in linux)

unreleased_history/public_api_changes/txn_get_attribute_group_iterator.md

@@ -1 +0,0 @@
-Added a new API `Transaction::GetAttributeGroupIterator` that can be used to create a multi-column-family attribute group iterator over the specified column families, including the data from both the transaction and the underlying database. This API is currently supported for optimistic and write-committed pessimistic transactions.

unreleased_history/public_api_changes/txn_get_coalescing_iterator.md

@@ -1 +0,0 @@
-Added a new API `Transaction::GetCoalescingIterator` that can be used to create a multi-column-family coalescing iterator over the specified column families, including the data from both the transaction and the underlying database. This API is currently supported for optimistic and write-committed pessimistic transactions.

util/file_reader_writer_test.cc

@@ -230,6 +230,98 @@ TEST_F(WritableFileWriterTest, IncrementalBuffer) {
   }
 }
 
+TEST_F(WritableFileWriterTest, AlignedBufferedWrites) {
+  class FakeWF : public FSWritableFile {
+   public:
+    explicit FakeWF(std::string* _file_data) : file_data_(_file_data) {}
+    ~FakeWF() override = default;
+
+    using FSWritableFile::Append;
+    IOStatus Append(const Slice& data, const IOOptions& /*options*/,
+                    IODebugContext* /*dbg*/) override {
+      EXPECT_EQ(data.size() & (data.size() - 1), 0);
+      file_data_->append(data.data(), data.size());
+      size_ += data.size();
+      return IOStatus::OK();
+    }
+    using FSWritableFile::PositionedAppend;
+    IOStatus PositionedAppend(const Slice& data, uint64_t pos,
+                              const IOOptions& /*options*/,
+                              IODebugContext* /*dbg*/) override {
+      EXPECT_TRUE(pos % 512 == 0);
+      EXPECT_TRUE(data.size() % 512 == 0);
+      file_data_->resize(pos);
+      file_data_->append(data.data(), data.size());
+      size_ += data.size();
+      return IOStatus::OK();
+    }
+
+    IOStatus Truncate(uint64_t size, const IOOptions& /*options*/,
+                      IODebugContext* /*dbg*/) override {
+      file_data_->resize(size);
+      return IOStatus::OK();
+    }
+    IOStatus Close(const IOOptions& /*options*/,
+                   IODebugContext* /*dbg*/) override {
+      return IOStatus::OK();
+    }
+    IOStatus Flush(const IOOptions& /*options*/,
+                   IODebugContext* /*dbg*/) override {
+      return IOStatus::OK();
+    }
+    IOStatus Sync(const IOOptions& /*options*/,
+                  IODebugContext* /*dbg*/) override {
+      return IOStatus::OK();
+    }
+    IOStatus Fsync(const IOOptions& /*options*/,
+                   IODebugContext* /*dbg*/) override {
+      return IOStatus::OK();
+    }
+    void SetIOPriority(Env::IOPriority /*pri*/) override {}
+    uint64_t GetFileSize(const IOOptions& /*options*/,
+                         IODebugContext* /*dbg*/) override {
+      return size_;
+    }
+    void GetPreallocationStatus(size_t* /*block_size*/,
+                                size_t* /*last_allocated_block*/) override {}
+    size_t GetUniqueId(char* /*id*/, size_t /*max_size*/) const override {
+      return 0;
+    }
+    IOStatus InvalidateCache(size_t /*offset*/, size_t /*length*/) override {
+      return IOStatus::OK();
+    }
+    bool use_direct_io() const override { return false; }
+
+    std::string* file_data_;
+    size_t size_ = 0;
+  };
+
+  Random r(301);
+  EnvOptions env_options;
+  env_options.writable_file_max_buffer_size = 64 * 1024 * 1024;
+  std::string actual;
+  std::unique_ptr<FakeWF> wf(new FakeWF(&actual));
+  std::unique_ptr<WritableFileWriter> writer(
+      new WritableFileWriter(std::move(wf), "" /* don't care */, env_options));
+
+  std::string target;
+  uint32_t left =
+      static_cast<uint32_t>(2 * env_options.writable_file_max_buffer_size);
+  ;
+  while (left > 0) {
+    uint32_t num = 4096 + r.Uniform(8192);
+    num = std::min<uint32_t>(num, left);
+    std::string random_string = r.RandomString(num);
+    ASSERT_OK(writer->Append(IOOptions(), Slice(random_string.c_str(), num)));
+    target.append(random_string.c_str(), num);
+    left -= num;
+  }
+  ASSERT_OK(writer->Flush(IOOptions()));
+  ASSERT_OK(writer->Close(IOOptions()));
+  ASSERT_EQ(target.size(), actual.size());
+  ASSERT_EQ(target, actual);
+}
+
 TEST_F(WritableFileWriterTest, BufferWithZeroCapacityDirectIO) {
   EnvOptions env_opts;
   env_opts.use_direct_writes = true;