[#27269] docdb: Add flag to allow continuing tablet writes after compaction failure

Summary:
Added `allow_compaction_failures_for_tablet_ids` tserver-side flag allowing to specify list of tablets for which we allow compactions to fail (for example due to on-disk data corruption) and that shoudn't fail subsequent writes to affected RocksDB instance. Should be used with extra care only for troubleshooting issues while/if we don't have other options to mitigate the problem.
Jira: DB-16750

Test Plan:
Manually tested with local cluster and sample workload:
```
./bin/yb-ctl destroy && ./bin/yb-ctl start

java -jar ~/code/yb-sample-apps/target/yb-sample-apps.jar --workload CassandraBatchTimeseries --nodes 127.0.0.1:9042 --num_threads_read 0 --num_threads_write 4 --nouuid --min_metrics_count 10000 --max_metrics_count 20000 --num_unique_keys 100000000

find ~/yugabyte-data -name "*.sst"

// cd to created tablet RocksDB directory

dd if=/dev/zero of=000010.sst.sblock.0 bs=1 conv=notrunc count=1000000 seek=100000

// Will crash on compaction

export TABLET_ID=...

./bin/yb-ctl restart --tserver_flags '"allow_compaction_failures_for_tablet_ids=$TABLET_ID"'

java -jar ~/code/yb-sample-apps/target/yb-sample-apps.jar --workload CassandraBatchTimeseries --nodes 127.0.0.1:9042 --num_threads_read 0 --num_threads_write 4 --nouuid --min_metrics_count 10000 --max_metrics_count 20000 --num_unique_keys 100000000

// Shouldn't crash on failed compactions, flushes should work

// Query accesing corrupt data should fail, but tserver shouldn't crash

./bin/ycqlsh --request-timeout 600 -e 'select count(*) from ybdemo_keyspace.batch_ts_metrics_raw'

./build/latest/bin/yb-ts-cli set_flag allow_compaction_failures_for_tablet_ids ""

// Should crash on next compaction attempt which includes 000010.sst

./bin/yb-ctl restart

./build/latest/bin/yb-ts-cli set_flag allow_compaction_failures_for_tablet_ids "$TABLET_ID"

java -jar ~/code/yb-sample-apps/target/yb-sample-apps.jar --workload CassandraBatchTimeseries --nodes 127.0.0.1:9042 --num_threads_read 0 --num_threads_write 4 --nouuid --min_metrics_count 10000 --max_metrics_count 20000 --num_unique_keys 100000000

// Shouldn't crash on failed compactions, flushes should work
```

Reviewers: hsunder, yyan, rthallam, arybochkin

Reviewed By: rthallam, arybochkin

Subscribers: yql, ybase

Tags: #jenkins-ready

Differential Revision: https://phorge.dev.yugabyte.com/D44079

Author: ttyusupov

src/yb/master/sys_catalog.cc

@@ -608,6 +608,7 @@ Status SysCatalogTable::OpenTablet(const scoped_refptr<tablet::RaftGroupMetadata
       .metric_registry = metric_registry_,
       .log_anchor_registry = tablet_peer()->log_anchor_registry(),
       .tablet_options = tablet_options,
+      .mutable_tablet_options = tablet::MutableTabletOptions{},
       .log_prefix_suffix = " P " + tablet_peer()->permanent_uuid(),
       .transaction_participant_context = tablet_peer().get(),
       .local_tablet_filter = client::LocalTabletFilter(),

src/yb/rocksdb/db.h

@@ -930,6 +930,12 @@ class DB {
   // Returns approximate middle key (see Version::GetMiddleKey).
   virtual yb::Result<std::string> GetMiddleKey() = 0;
 
+  // If true, will allow compactions to fail without setting bg_error and not causing writes to
+  // fail. Should only be used with extra care for troubleshooting when/while there are no other
+  // options available.
+  // Default: false
+  virtual void SetAllowCompactionFailures(AllowCompactionFailures allow_compaction_failures) = 0;
+
   // Returns a table reader for the largest SST file.
   virtual yb::Result<TableReader*> TEST_GetLargestSstTableReader() {
     return STATUS(NotSupported, "");

src/yb/rocksdb/db/db_impl.cc

@@ -2603,7 +2603,8 @@ Status DBImpl::CompactFilesImpl(
         "[%s] [JOB %d] Compaction error: %s",
         c->column_family_data()->GetName().c_str(), job_context->job_id,
         status.ToString().c_str());
-    if (db_options_.paranoid_checks && bg_error_.ok()) {
+    if (db_options_.paranoid_checks && !allow_compaction_failures_ &&
+        bg_error_.ok()) {
       bg_error_ = status;
     }
   }
@@ -4046,7 +4047,8 @@ Result<FileNumbersHolder> DBImpl::BackgroundCompaction(
   } else {
     RLOG(InfoLogLevel::WARN_LEVEL, db_options_.info_log, "Compaction error: %s",
         status.ToString().c_str());
-    if (db_options_.paranoid_checks && bg_error_.ok()) {
+    if (db_options_.paranoid_checks && !allow_compaction_failures_ &&
+        bg_error_.ok()) {
       bg_error_ = status;
     }
   }
@@ -7140,4 +7142,10 @@ size_t DBImpl::TEST_NumNotStartedCompactionsUnlocked(CompactionSizeKind compacti
       });
 }
 
+void DBImpl::SetAllowCompactionFailures(AllowCompactionFailures allow_compaction_failures) {
+  if (allow_compaction_failures_.exchange(allow_compaction_failures) != allow_compaction_failures) {
+    LOG_WITH_PREFIX(INFO) << "allow_compaction_failures changed to: " << allow_compaction_failures;
+  }
+}
+
 }  // namespace rocksdb

src/yb/rocksdb/db/db_impl.h

@@ -508,6 +508,8 @@ class DBImpl : public DB {
 
   Result<std::string> GetMiddleKey() override;
 
+  void SetAllowCompactionFailures(AllowCompactionFailures allow_compaction_failures) override;
+
   // Returns a table reader for the largest SST file.
   Result<TableReader*> TEST_GetLargestSstTableReader() override;
 
@@ -1027,6 +1029,8 @@ class DBImpl : public DB {
 
   std::function<void()> files_changed_listener_ GUARDED_BY(files_changed_listener_mutex_);
 
+  std::atomic<bool> allow_compaction_failures_{false};
+
   // No copying allowed
   DBImpl(const DBImpl&) = delete;
   void operator=(const DBImpl&) = delete;

src/yb/rocksdb/db/db_test.cc

@@ -4754,6 +4754,10 @@ class ModelDB: public DB {
     return NotSupported();
   }
 
+  void SetAllowCompactionFailures(AllowCompactionFailures allow_compaction_failures) override {
+    LOG(FATAL) << "SetAllowCompactionFailures is not supported.";
+  }
+
  private:
   Status NotSupported() const {
     return STATUS(NotSupported, "Not supported in Model DB");

src/yb/rocksdb/rocksdb_fwd.h

@@ -68,6 +68,8 @@ class DirectWriteHandler;
 
 YB_DEFINE_ENUM(FlushAbility, (kNoNewData)(kHasNewData)(kAlreadyFlushing));
 
+YB_STRONGLY_TYPED_BOOL(AllowCompactionFailures);
+
 // Frontier should be copyable, but should still preserve its polymorphic nature. We cannot use
 // shared_ptr here, because we are planning to modify the copied value. If we used shared_ptr and
 // modified the copied value, the original value would also change.

src/yb/tablet/tablet-harness.cc

@@ -103,6 +103,7 @@ TabletInitData TabletHarness::MakeTabletInitData(const RaftGroupMetadataPtr& met
       .metric_registry = metrics_registry_.get(),
       .log_anchor_registry = new log::LogAnchorRegistry(),
       .tablet_options = TabletOptions(),
+      .mutable_tablet_options = MutableTabletOptions{},
       .log_prefix_suffix = std::string(),
       .transaction_participant_context = nullptr,
       .local_tablet_filter = client::LocalTabletFilter(),

src/yb/tablet/tablet.cc

@@ -649,6 +649,7 @@ Tablet::Tablet(const TabletInitData& data)
       mvcc_(
           MakeTabletLogPrefix(data.metadata->raft_group_id(), data.log_prefix_suffix), data.clock),
       tablet_options_(data.tablet_options),
+      mutable_tablet_options_(data.mutable_tablet_options),
       pending_op_counter_blocking_rocksdb_shutdown_start_(Format(
           "T $0 Read/write operations blocking start of RocksDB shutdown",
           metadata_->raft_group_id())),
@@ -1007,6 +1008,18 @@ Status Tablet::OpenKeyValueTablet() {
   RETURN_NOT_OK(snapshots_->Open());
   RETURN_NOT_OK(OpenRegularDB(common_options));
   RETURN_NOT_OK(OpenIntentsDB(common_options));
+  rocksdb::AllowCompactionFailures allow_compaction_failures =
+      rocksdb::AllowCompactionFailures::kFalse;
+  {
+    SharedLock lock(mutable_tablet_options_mutex_);
+    allow_compaction_failures = mutable_tablet_options_.allow_compaction_failures;
+  }
+  for (auto* db : {regular_db_.get(), intents_db_.get()}) {
+    if (!db) {
+      continue;
+    }
+    db->SetAllowCompactionFailures(allow_compaction_failures);
+  }
 
   // Don't allow reads at timestamps lower than the highest history cutoff of a past compaction.
   auto regular_flushed_frontier = regular_db_->GetFlushedFrontier();
@@ -5553,6 +5566,21 @@ void Tablet::SetCompactFlushRateLimitBytesPerSec(int64_t bytes_per_sec) {
   }
 }
 
+void Tablet::SetAllowCompactionFailures(
+    rocksdb::AllowCompactionFailures allow_compaction_failures) {
+  {
+    std::lock_guard lock(mutable_tablet_options_mutex_);
+    mutable_tablet_options_.allow_compaction_failures = allow_compaction_failures;
+  }
+  auto scoped_op = CreateScopedRWOperationBlockingRocksDbShutdownStart();
+  for (auto* db : {regular_db(), intents_db()}) {
+    if (!db) {
+      continue;
+    }
+    db->SetAllowCompactionFailures(allow_compaction_failures);
+  }
+}
+
 // ------------------------------------------------------------------------------------------------
 
 Result<ScopedReadOperation> ScopedReadOperation::Create(

src/yb/tablet/tablet.h

@@ -993,6 +993,8 @@ class Tablet : public AbstractTablet,
   void RefreshCompactFlushRateLimitBytesPerSec();
   void SetCompactFlushRateLimitBytesPerSec(int64_t bytes_per_sec);
 
+  void SetAllowCompactionFailures(rocksdb::AllowCompactionFailures allow_compaction_failures);
+
  private:
   friend class Iterator;
   friend class TabletPeerTest;
@@ -1153,6 +1155,9 @@ class Tablet : public AbstractTablet,
   // For the block cache and memory manager shared across tablets
   const TabletOptions tablet_options_;
 
+  mutable std::shared_mutex mutable_tablet_options_mutex_;
+  MutableTabletOptions mutable_tablet_options_ GUARDED_BY (mutable_tablet_options_mutex_);
+
   // A lightweight way to reject new operations when the tablet is shutting down. This is used to
   // prevent race conditions between destructing the RocksDB in-memory instance and read/write
   // operations.

src/yb/tablet/tablet_bootstrap-test.cc

@@ -233,6 +233,7 @@ class BootstrapTest : public LogTestBase {
       .metric_registry = nullptr,
       .log_anchor_registry = log_anchor_registry,
       .tablet_options = tablet_options,
+      .mutable_tablet_options = MutableTabletOptions{},
       .log_prefix_suffix = std::string(),
       .transaction_participant_context = nullptr,
       .local_tablet_filter = client::LocalTabletFilter(),

src/yb/tablet/tablet_options.h

@@ -29,6 +29,8 @@
 
 #include "yb/docdb/local_waiting_txn_registry.h"
 
+#include "yb/rocksdb/rocksdb_fwd.h"
+
 #include "yb/server/server_fwd.h"
 
 #include "yb/tablet/tablet_fwd.h"
@@ -64,6 +66,11 @@ struct TabletOptions {
   std::shared_ptr<rocksdb::RocksDBPriorityThreadPoolMetrics> priority_thread_pool_metrics;
 };
 
+struct MutableTabletOptions {
+  rocksdb::AllowCompactionFailures allow_compaction_failures =
+      rocksdb::AllowCompactionFailures::kFalse;
+};
+
 using TransactionManagerProvider = std::function<client::TransactionManager&()>;
 
 YB_DEFINE_ENUM(VectorIndexThreadPoolType, (kBackground)(kBackfill)(kInsert));
@@ -81,7 +88,8 @@ struct TabletInitData {
   std::shared_ptr<MemTracker> block_based_table_mem_tracker;
   MetricRegistry* metric_registry = nullptr;
   log::LogAnchorRegistryPtr log_anchor_registry;
-  const TabletOptions tablet_options;
+  TabletOptions tablet_options;
+  MutableTabletOptions mutable_tablet_options;
   std::string log_prefix_suffix;
   TransactionParticipantContext* transaction_participant_context = nullptr;
   client::LocalTabletFilter local_tablet_filter;

src/yb/tools/fs_tool.cc

@@ -272,6 +272,7 @@ Status FsTool::DumpTabletData(const std::string& tablet_id) {
       .metric_registry = nullptr,
       .log_anchor_registry = reg.get(),
       .tablet_options = tablet_options,
+      .mutable_tablet_options = tablet::MutableTabletOptions{},
       .log_prefix_suffix = std::string(),
       .transaction_participant_context = nullptr,
       .local_tablet_filter = client::LocalTabletFilter(),

src/yb/tools/sys-catalog-tool.cc

@@ -205,6 +205,7 @@ Status MiniSysCatalogTable::OpenTablet(const scoped_refptr<tablet::RaftGroupMeta
       .metric_registry = metric_registry_.get(),
       .log_anchor_registry = nullptr,
       .tablet_options = tablet_options_,
+      .mutable_tablet_options = tablet::MutableTabletOptions{},
       .log_prefix_suffix = "",
       .transaction_participant_context = tablet_peer_.get(),
       .local_tablet_filter = nullptr,

src/yb/tserver/ts_tablet_manager.cc

@@ -298,6 +298,12 @@ DEFINE_RUNTIME_bool(enable_copy_retryable_requests_from_parent, true,
                     "Whether to copy retryable requests from parent tablet when opening"
                     "the child tablet");
 
+DEFINE_RUNTIME_string(allow_compaction_failures_for_tablet_ids, "",
+    "List of tablet IDs for which compaction failures are allowed and will not cause write "
+    "failures and FATALs.");
+TAG_FLAG(allow_compaction_failures_for_tablet_ids, hidden);
+TAG_FLAG(allow_compaction_failures_for_tablet_ids, advanced);
+
 DEFINE_NON_RUNTIME_uint32(deleted_tablet_cache_max_size, 10000,
                           "Maximum size for the cache of recently deleted tablet ids. Used to "
                           "reject remote bootstrap requests for recently deleted tablets.");
@@ -629,10 +635,14 @@ Status TSTabletManager::Init() {
     tablet_options_.rate_limiter = docdb::CreateRocksDBRateLimiter();
   }
 
-  rate_limiter_flag_callback_ = CHECK_RESULT(RegisterFlagUpdateCallback(
+  flag_callbacks_.emplace_back(VERIFY_RESULT(RegisterFlagUpdateCallback(
+      &FLAGS_allow_compaction_failures_for_tablet_ids,
+      "allow_compaction_failures_for_tablet_ids",
+      [this] { UpdateAllowCompactionFailures(); })));
+  flag_callbacks_.emplace_back(VERIFY_RESULT(RegisterFlagUpdateCallback(
       &FLAGS_rocksdb_compact_flush_rate_limit_bytes_per_sec,
       "RocksDBCompactFlushRateLimiter",
-      [this] { UpdateCompactFlushRateLimitBytesPerSec(); }));
+      [this] { UpdateCompactFlushRateLimitBytesPerSec(); })));
 
   // Start the threadpool we'll use to open tablets.
   // This has to be done in Init() instead of the constructor, since the
@@ -771,6 +781,11 @@ Status TSTabletManager::Init() {
 
   {
     std::lock_guard lock(mutex_);
+    allow_compaction_failures_for_tablet_ids_ = FLAGS_allow_compaction_failures_for_tablet_ids;
+    if (!allow_compaction_failures_for_tablet_ids_.empty()) {
+      LOG_WITH_PREFIX(INFO) << "Flag allow_compaction_failures_for_tablet_ids is set to: "
+                            << allow_compaction_failures_for_tablet_ids_;
+    }
     state_ = MANAGER_RUNNING;
   }
 
@@ -2058,6 +2073,14 @@ void TSTabletManager::OpenTablet(const RaftGroupMetadataPtr& meta,
     }
     TEST_PAUSE_IF_FLAG(TEST_pause_after_set_bootstrapping);
 
+    rocksdb::AllowCompactionFailures allow_compaction_failures =
+        rocksdb::AllowCompactionFailures::kFalse;
+    {
+      SharedLock lock(mutex_);
+      allow_compaction_failures = rocksdb::AllowCompactionFailures(
+          allow_compaction_failures_for_tablet_ids_.contains(tablet_id));
+    }
+
     tablet::TabletInitData tablet_init_data = {
         .metadata = meta,
         .client_future = server_->client_future(),
@@ -2067,6 +2090,9 @@ void TSTabletManager::OpenTablet(const RaftGroupMetadataPtr& meta,
         .metric_registry = metric_registry_,
         .log_anchor_registry = tablet_peer->log_anchor_registry(),
         .tablet_options = tablet_options_,
+        .mutable_tablet_options = {
+            .allow_compaction_failures = allow_compaction_failures,
+        },
         .log_prefix_suffix = " P " + tablet_peer->permanent_uuid(),
         .transaction_participant_context = tablet_peer.get(),
         .local_tablet_filter = std::bind(&TSTabletManager::PreserveLocalLeadersOnly, this, _1),
@@ -2257,7 +2283,10 @@ void TSTabletManager::StartShutdown() {
     }
   }
 
-  rate_limiter_flag_callback_.Deregister();
+  for (auto& callback : flag_callbacks_) {
+    callback.Deregister();
+  }
+  flag_callbacks_.clear();
 
   {
     std::lock_guard lock(service_registration_mutex_);
@@ -3534,6 +3563,26 @@ void TSTabletManager::UpdateCompactFlushRateLimitBytesPerSec() {
   }
 }
 
+void TSTabletManager::UpdateAllowCompactionFailures() {
+  std::string allow_compaction_failures_for_tablet_ids;
+  {
+    std::lock_guard lock(mutex_);
+    allow_compaction_failures_for_tablet_ids_ = FLAGS_allow_compaction_failures_for_tablet_ids;
+    allow_compaction_failures_for_tablet_ids = allow_compaction_failures_for_tablet_ids_;
+  }
+  for (const auto& tablet_peer : GetTabletPeers()) {
+    const auto shared_tablet = tablet_peer->shared_tablet();
+    if (!shared_tablet) {
+      continue;
+    }
+
+    const auto allow_compaction_failures = rocksdb::AllowCompactionFailures(
+        allow_compaction_failures_for_tablet_ids.contains(shared_tablet->tablet_id()));
+
+    shared_tablet->SetAllowCompactionFailures(allow_compaction_failures);
+  }
+}
+
 rpc::ThreadPool* TSTabletManager::VectorIndexThreadPool(tablet::VectorIndexThreadPoolType type) {
   auto& thread_pool_ptr = vector_index_thread_pools_[std::to_underlying(type)];
   auto result = thread_pool_ptr.get();

src/yb/tserver/ts_tablet_manager.h

@@ -644,6 +644,8 @@ class TSTabletManager : public tserver::TabletPeerLookupIf, public tablet::Table
 
   void UpdateCompactFlushRateLimitBytesPerSec();
 
+  void UpdateAllowCompactionFailures();
+
   rpc::ThreadPool* VectorIndexThreadPool(tablet::VectorIndexThreadPoolType type);
   PriorityThreadPool* VectorIndexPriorityThreadPool(tablet::VectorIndexPriorityThreadPoolType type);
 
@@ -822,8 +824,9 @@ class TSTabletManager : public tserver::TabletPeerLookupIf, public tablet::Table
   std::shared_ptr<client::YBMetaDataCache> metadata_cache_holder_;
   std::atomic<client::YBMetaDataCache*> metadata_cache_;
 
-  // Callback for FLAGS_rocksdb_compact_flush_rate_limit_bytes_per_sec update handling.
-  FlagCallbackRegistration rate_limiter_flag_callback_;
+  std::vector<FlagCallbackRegistration> flag_callbacks_;
+
+  std::string allow_compaction_failures_for_tablet_ids_ GUARDED_BY(mutex_);
 
   std::mutex vector_index_thread_pool_mutex_;
   std::array<AtomicUniquePtr<rpc::ThreadPool>, tablet::kVectorIndexThreadPoolTypeMapSize>