Add missing db crash options (#12414)

Summary:
**Context/Summary:**
We are doing a sweep in all public options, including but not limited to the `Options`, `Read/WriteOptions`, `IngestExternalFileOptions`, cache options.., to find and add the uncovered ones into db crash. The options included in this PR require minimum changes to db crash other than adding the options themselves.

A bonus change: to surface new issues by improved coverage in stderror, we decided to fail/terminate crash test for manual compactions (CompactFiles, CompactRange()) on meaningful errors. See https://github.com/facebook/rocksdb/pull/12414/files#diff-5c4ced6afb6a90e27fec18ab03b2cd89e8f99db87791b4ecc6fa2694284d50c0R2528-R2532, https://github.com/facebook/rocksdb/pull/12414/files#diff-5c4ced6afb6a90e27fec18ab03b2cd89e8f99db87791b4ecc6fa2694284d50c0R2330-R2336 for more.

Pull Request resolved: https://github.com/facebook/rocksdb/pull/12414

Test Plan:
- Run `python3 ./tools/db_crashtest.py --simple blackbox` for 10 minutes to ensure no trivial failure
- Run `python3 tools/db_crashtest.py --simple blackbox --compact_files_one_in=1 --compact_range_one_in=1 --read_fault_one_in=1 --write_fault_one_in=1 --interval=50` for a while to ensure the bonus change does not result in trivial crash/termination of stress test

Reviewed By: ajkr, jowlyzhang, cbi42

Differential Revision: D54691774

Pulled By: hx235

fbshipit-source-id: 50443dfb6aaabd8e24c79a2e42b68c6de877be88

5 files changed, 454 insertions, 47 deletions

diff --git a/db_stress_tool/db_stress_common.h b/db_stress_tool/db_stress_common.h
index 768e90b9f..977aecff6 100644
--- a/db_stress_tool/db_stress_common.h
+++ b/db_stress_tool/db_stress_common.h
@@ -352,6 +352,45 @@ DECLARE_uint64(initial_auto_readahead_size);
 DECLARE_uint64(max_auto_readahead_size);
 DECLARE_uint64(num_file_reads_for_auto_readahead);
 DECLARE_bool(auto_readahead_size);
+DECLARE_bool(allow_fallocate);
+DECLARE_int32(table_cache_numshardbits);
+DECLARE_bool(enable_write_thread_adaptive_yield);
+DECLARE_uint64(log_readahead_size);
+DECLARE_uint64(bgerror_resume_retry_interval);
+DECLARE_uint64(delete_obsolete_files_period_micros);
+DECLARE_uint64(max_log_file_size);
+DECLARE_uint64(log_file_time_to_roll);
+DECLARE_bool(use_adaptive_mutex);
+DECLARE_bool(advise_random_on_open);
+DECLARE_uint64(WAL_ttl_seconds);
+DECLARE_uint64(WAL_size_limit_MB);
+DECLARE_bool(strict_bytes_per_sync);
+DECLARE_bool(avoid_flush_during_shutdown);
+DECLARE_bool(fill_cache);
+DECLARE_bool(optimize_multiget_for_io);
+DECLARE_bool(memtable_insert_hint_per_batch);
+DECLARE_bool(dump_malloc_stats);
+DECLARE_uint64(stats_history_buffer_size);
+DECLARE_bool(skip_stats_update_on_db_open);
+DECLARE_bool(optimize_filters_for_hits);
+DECLARE_uint64(sample_for_compression);
+DECLARE_bool(report_bg_io_stats);
+DECLARE_bool(cache_index_and_filter_blocks_with_high_priority);
+DECLARE_bool(use_delta_encoding);
+DECLARE_bool(verify_compression);
+DECLARE_uint32(read_amp_bytes_per_bit);
+DECLARE_bool(enable_index_compression);
+DECLARE_uint32(index_shortening);
+DECLARE_uint32(metadata_charge_policy);
+DECLARE_bool(use_adaptive_mutex_lru);
+DECLARE_uint32(compress_format_version);
+DECLARE_uint64(manifest_preallocation_size);
+DECLARE_bool(enable_checksum_handoff);
+DECLARE_uint64(max_total_wal_size);
+DECLARE_double(high_pri_pool_ratio);
+DECLARE_double(low_pri_pool_ratio);
+DECLARE_uint64(soft_pending_compaction_bytes_limit);
+DECLARE_uint64(hard_pending_compaction_bytes_limit);
 
 constexpr long KB = 1024;
 constexpr int kRandomValueMaxFactor = 3;
diff --git a/db_stress_tool/db_stress_gflags.cc b/db_stress_tool/db_stress_gflags.cc
index 3a0059425..3156cd2d2 100644
--- a/db_stress_tool/db_stress_gflags.cc
+++ b/db_stress_tool/db_stress_gflags.cc
@@ -7,6 +7,9 @@
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file. See the AUTHORS file for names of contributors.
 
+#include "rocksdb/cache.h"
+#include "rocksdb/options.h"
+#include "rocksdb/utilities/backup_engine.h"
 #ifdef GFLAGS
 #include "db_stress_tool/db_stress_common.h"
 
@@ -402,7 +405,8 @@ DEFINE_double(experimental_mempurge_threshold, 0.0,
               "Maximum estimated useful payload that triggers a "
               "mempurge process to collect memtable garbage bytes.");
 
-DEFINE_bool(enable_write_thread_adaptive_yield, true,
+DEFINE_bool(enable_write_thread_adaptive_yield,
+            ROCKSDB_NAMESPACE::Options().enable_write_thread_adaptive_yield,
             "Use a yielding spin loop for brief writer thread waits.");
 
 // Options for StackableDB-based BlobDB
@@ -1144,4 +1148,155 @@ DEFINE_uint32(bottommost_file_compaction_delay, 0,
 DEFINE_bool(auto_readahead_size, false,
             "Does auto tuning of readahead_size when enabled during scans.");
 
+DEFINE_bool(allow_fallocate, ROCKSDB_NAMESPACE::Options().allow_fallocate,
+            "Options.allow_fallocate");
+
+DEFINE_int32(table_cache_numshardbits,
+             ROCKSDB_NAMESPACE::Options().table_cache_numshardbits,
+             "Options.table_cache_numshardbits");
+
+DEFINE_uint64(log_readahead_size,
+              ROCKSDB_NAMESPACE::Options().log_readahead_size,
+              "Options.log_readahead_size");
+
+DEFINE_uint64(bgerror_resume_retry_interval,
+              ROCKSDB_NAMESPACE::Options().bgerror_resume_retry_interval,
+              "Options.bgerror_resume_retry_interval");
+
+DEFINE_uint64(delete_obsolete_files_period_micros,
+              ROCKSDB_NAMESPACE::Options().delete_obsolete_files_period_micros,
+              "Options.delete_obsolete_files_period_micros");
+
+DEFINE_uint64(max_log_file_size, ROCKSDB_NAMESPACE::Options().max_log_file_size,
+              "Options.max_log_file_sizes");
+
+DEFINE_uint64(log_file_time_to_roll,
+              ROCKSDB_NAMESPACE::Options().log_file_time_to_roll,
+              "Options.log_file_time_to_roll");
+
+DEFINE_bool(use_adaptive_mutex, ROCKSDB_NAMESPACE::Options().use_adaptive_mutex,
+            "Options.use_adaptive_mutex");
+
+DEFINE_bool(advise_random_on_open,
+            ROCKSDB_NAMESPACE::Options().advise_random_on_open,
+            "Options.advise_random_on_open");
+
+DEFINE_uint64(WAL_ttl_seconds, ROCKSDB_NAMESPACE::Options().WAL_ttl_seconds,
+              "Options.WAL_ttl_seconds");
+
+DEFINE_uint64(WAL_size_limit_MB, ROCKSDB_NAMESPACE::Options().WAL_size_limit_MB,
+              "Options.WAL_size_limit_MB");
+
+DEFINE_bool(strict_bytes_per_sync,
+            ROCKSDB_NAMESPACE::Options().strict_bytes_per_sync,
+            "Options.strict_bytes_per_sync");
+
+DEFINE_bool(avoid_flush_during_shutdown,
+            ROCKSDB_NAMESPACE::Options().avoid_flush_during_shutdown,
+            "Options.avoid_flush_during_shutdown");
+
+DEFINE_bool(fill_cache, ROCKSDB_NAMESPACE::ReadOptions().fill_cache,
+            "ReadOptions.fill_cache");
+
+DEFINE_bool(optimize_multiget_for_io,
+            ROCKSDB_NAMESPACE::ReadOptions().optimize_multiget_for_io,
+            "ReadOptions.optimize_multiget_for_io");
+
+DEFINE_bool(memtable_insert_hint_per_batch,
+            ROCKSDB_NAMESPACE::WriteOptions().memtable_insert_hint_per_batch,
+            "WriteOptions.memtable_insert_hint_per_batch");
+
+DEFINE_bool(dump_malloc_stats, ROCKSDB_NAMESPACE::Options().dump_malloc_stats,
+            "Options.dump_malloc_stats");
+
+DEFINE_uint64(stats_history_buffer_size,
+              ROCKSDB_NAMESPACE::Options().stats_history_buffer_size,
+              "Options.stats_history_buffer_size");
+
+DEFINE_bool(skip_stats_update_on_db_open,
+            ROCKSDB_NAMESPACE::Options().skip_stats_update_on_db_open,
+            "Options.skip_stats_update_on_db_open");
+
+DEFINE_bool(optimize_filters_for_hits,
+            ROCKSDB_NAMESPACE::Options().optimize_filters_for_hits,
+            "Options.optimize_filters_for_hits");
+
+DEFINE_uint64(sample_for_compression,
+              ROCKSDB_NAMESPACE::Options().sample_for_compression,
+              "Options.sample_for_compression");
+
+DEFINE_bool(report_bg_io_stats, ROCKSDB_NAMESPACE::Options().report_bg_io_stats,
+            "Options.report_bg_io_stats");
+
+DEFINE_bool(
+    cache_index_and_filter_blocks_with_high_priority,
+    ROCKSDB_NAMESPACE::BlockBasedTableOptions()
+        .cache_index_and_filter_blocks_with_high_priority,
+    "BlockBasedTableOptions.cache_index_and_filter_blocks_with_high_priority");
+
+DEFINE_bool(use_delta_encoding,
+            ROCKSDB_NAMESPACE::BlockBasedTableOptions().use_delta_encoding,
+            "BlockBasedTableOptions.use_delta_encoding");
+
+DEFINE_bool(verify_compression,
+            ROCKSDB_NAMESPACE::BlockBasedTableOptions().verify_compression,
+            "BlockBasedTableOptions.verify_compression");
+
+DEFINE_uint32(
+    read_amp_bytes_per_bit,
+    ROCKSDB_NAMESPACE::BlockBasedTableOptions().read_amp_bytes_per_bit,
+    "Options.read_amp_bytes_per_bit");
+
+DEFINE_bool(
+    enable_index_compression,
+    ROCKSDB_NAMESPACE::BlockBasedTableOptions().enable_index_compression,
+    "BlockBasedTableOptions.enable_index_compression");
+
+DEFINE_uint32(index_shortening,
+              static_cast<uint32_t>(
+                  ROCKSDB_NAMESPACE::BlockBasedTableOptions().index_shortening),
+              "BlockBasedTableOptions.index_shortening");
+
+DEFINE_uint32(metadata_charge_policy,
+              static_cast<uint32_t>(ROCKSDB_NAMESPACE::ShardedCacheOptions()
+                                        .metadata_charge_policy),
+              "ShardedCacheOptions.metadata_charge_policy");
+
+DEFINE_bool(use_adaptive_mutex_lru,
+            ROCKSDB_NAMESPACE::LRUCacheOptions().use_adaptive_mutex,
+            "LRUCacheOptions.use_adaptive_mutex");
+
+DEFINE_uint32(
+    compress_format_version,
+    static_cast<uint32_t>(ROCKSDB_NAMESPACE::CompressedSecondaryCacheOptions()
+                              .compress_format_version),
+    "CompressedSecondaryCacheOptions.compress_format_version");
+
+DEFINE_uint64(manifest_preallocation_size,
+              ROCKSDB_NAMESPACE::Options().manifest_preallocation_size,
+              "Options.manifest_preallocation_size");
+
+DEFINE_uint64(max_total_wal_size,
+              ROCKSDB_NAMESPACE::Options().max_total_wal_size,
+              "Options.max_total_wal_size");
+
+DEFINE_bool(enable_checksum_handoff, false,
+            "If true, include all the supported files in "
+            "Options.checksum_handoff_file. Otherwise include no files.");
+
+DEFINE_double(high_pri_pool_ratio,
+              ROCKSDB_NAMESPACE::LRUCacheOptions().high_pri_pool_ratio,
+              "LRUCacheOptions.high_pri_pool_ratio");
+
+DEFINE_double(low_pri_pool_ratio,
+              ROCKSDB_NAMESPACE::LRUCacheOptions().low_pri_pool_ratio,
+              "LRUCacheOptions.low_pri_pool_ratio");
+
+DEFINE_uint64(soft_pending_compaction_bytes_limit,
+              ROCKSDB_NAMESPACE::Options().soft_pending_compaction_bytes_limit,
+              "Options.soft_pending_compaction_bytes_limit");
+
+DEFINE_uint64(hard_pending_compaction_bytes_limit,
+              ROCKSDB_NAMESPACE::Options().hard_pending_compaction_bytes_limit,
+              "Options.hard_pending_compaction_bytes_limit");
 #endif  // GFLAGS
diff --git a/db_stress_tool/db_stress_test_base.cc b/db_stress_tool/db_stress_test_base.cc
index feec339d9..7376defbb 100644
--- a/db_stress_tool/db_stress_test_base.cc
+++ b/db_stress_tool/db_stress_test_base.cc
@@ -141,6 +141,7 @@ std::shared_ptr<Cache> StressTest::NewCache(size_t capacity,
     }
     CompressedSecondaryCacheOptions opts;
     opts.capacity = FLAGS_compressed_secondary_cache_size;
+    opts.compress_format_version = FLAGS_compress_format_version;
     secondary_cache = NewCompressedSecondaryCache(opts);
     if (secondary_cache == nullptr) {
       fprintf(stderr, "Failed to allocate compressed secondary cache\n");
@@ -191,6 +192,11 @@ std::shared_ptr<Cache> StressTest::NewCache(size_t capacity,
     LRUCacheOptions opts;
     opts.capacity = capacity;
     opts.num_shard_bits = num_shard_bits;
+    opts.metadata_charge_policy =
+        static_cast<CacheMetadataChargePolicy>(FLAGS_metadata_charge_policy);
+    opts.use_adaptive_mutex = FLAGS_use_adaptive_mutex_lru;
+    opts.high_pri_pool_ratio = FLAGS_high_pri_pool_ratio;
+    opts.low_pri_pool_ratio = FLAGS_low_pri_pool_ratio;
     if (tiered) {
       TieredCacheOptions tiered_opts;
       tiered_opts.cache_opts = &opts;
@@ -830,10 +836,14 @@ void StressTest::OperateDb(ThreadState* thread) {
   read_opts.adaptive_readahead = FLAGS_adaptive_readahead;
   read_opts.readahead_size = FLAGS_readahead_size;
   read_opts.auto_readahead_size = FLAGS_auto_readahead_size;
+  read_opts.fill_cache = FLAGS_fill_cache;
+  read_opts.optimize_multiget_for_io = FLAGS_optimize_multiget_for_io;
   WriteOptions write_opts;
   if (FLAGS_rate_limit_auto_wal_flush) {
     write_opts.rate_limiter_priority = Env::IO_USER;
   }
+  write_opts.memtable_insert_hint_per_batch =
+      FLAGS_memtable_insert_hint_per_batch;
   auto shared = thread->shared;
   char value[100];
   std::string from_db;
@@ -1264,6 +1274,10 @@ Status StressTest::TestIterate(ThreadState* thread,
   Slice read_ts_slice;
   MaybeUseOlderTimestampForRangeScan(thread, read_ts_str, read_ts_slice, ro);
 
+  std::string op_logs;
+  ro.pin_data = thread->rand.OneIn(2);
+  ro.background_purge_on_iterator_cleanup = thread->rand.OneIn(2);
+
   bool expect_total_order = false;
   if (thread->rand.OneIn(16)) {
     // When prefix extractor is used, it's useful to cover total order seek.
@@ -1315,7 +1329,6 @@ Status StressTest::TestIterate(ThreadState* thread,
     }
   }
 
-  std::string op_logs;
   constexpr size_t kOpLogsLimit = 10000;
 
   for (const std::string& key_str : key_strs) {
@@ -1343,18 +1356,6 @@ Status StressTest::TestIterate(ThreadState* thread,
       lower_bound = Slice(lower_bound_str);
     }
 
-    // Record some options to op_logs
-    op_logs += "total_order_seek: ";
-    op_logs += (ro.total_order_seek ? "1 " : "0 ");
-    op_logs += "auto_prefix_mode: ";
-    op_logs += (ro.auto_prefix_mode ? "1 " : "0 ");
-    if (ro.iterate_upper_bound != nullptr) {
-      op_logs += "ub: " + upper_bound.ToString(true) + " ";
-    }
-    if (ro.iterate_lower_bound != nullptr) {
-      op_logs += "lb: " + lower_bound.ToString(true) + " ";
-    }
-
     // Set up an iterator, perform the same operations without bounds and with
     // total order seek, and compare the results. This is to identify bugs
     // related to bounds, prefix extractor, or reseeking. Sometimes we are
@@ -1531,7 +1532,20 @@ void StressTest::VerifyIterator(ThreadState* thread,
                                  ? nullptr
                                  : options_.prefix_extractor.get();
   const Comparator* cmp = options_.comparator;
-
+  std::ostringstream read_opt_oss;
+  read_opt_oss << "pin_data: " << ro.pin_data
+               << ", background_purge_on_iterator_cleanup: "
+               << ro.background_purge_on_iterator_cleanup
+               << ", total_order_seek: " << ro.total_order_seek
+               << ", auto_prefix_mode: " << ro.auto_prefix_mode
+               << ", iterate_upper_bound: "
+               << (ro.iterate_upper_bound
+                       ? ro.iterate_upper_bound->ToString(true).c_str()
+                       : "")
+               << ", iterate_lower_bound: "
+               << (ro.iterate_lower_bound
+                       ? ro.iterate_lower_bound->ToString(true).c_str()
+                       : "");
   if (iter->Valid() && !cmp_iter->Valid()) {
     if (pe != nullptr) {
       if (!pe->InDomain(seek_key)) {
@@ -1550,8 +1564,10 @@ void StressTest::VerifyIterator(ThreadState* thread,
     }
     fprintf(stderr,
             "Control iterator is invalid but iterator has key %s "
-            "%s\n",
-            iter->key().ToString(true).c_str(), op_logs.c_str());
+            "%s under specified iterator ReadOptions: %s (Empty string or "
+            "missing field indicates default option or value is used)\n",
+            iter->key().ToString(true).c_str(), op_logs.c_str(),
+            read_opt_oss.str().c_str());
 
     *diverged = true;
   } else if (cmp_iter->Valid()) {
@@ -1579,9 +1595,12 @@ void StressTest::VerifyIterator(ThreadState* thread,
         }
         fprintf(stderr,
                 "Iterator stays in prefix but control doesn't"
-                " iterator key %s control iterator key %s %s\n",
+                " iterator key %s control iterator key %s %s under specified "
+                "iterator ReadOptions: %s (Empty string or "
+                "missing field indicates default option or value is used)\n",
                 iter->key().ToString(true).c_str(),
-                cmp_iter->key().ToString(true).c_str(), op_logs.c_str());
+                cmp_iter->key().ToString(true).c_str(), op_logs.c_str(),
+                read_opt_oss.str().c_str());
       }
     }
     // Check upper or lower bounds.
@@ -1598,8 +1617,11 @@ void StressTest::VerifyIterator(ThreadState* thread,
                                          /*b_has_ts=*/false) > 0))) {
         fprintf(stderr,
                 "Iterator diverged from control iterator which"
-                " has value %s %s\n",
-                total_order_key.ToString(true).c_str(), op_logs.c_str());
+                " has value %s %s  under specified iterator ReadOptions: %s "
+                "(Empty string or "
+                "missing field indicates default option or value is used)\n",
+                total_order_key.ToString(true).c_str(), op_logs.c_str(),
+                read_opt_oss.str().c_str());
         if (iter->Valid()) {
           fprintf(stderr, "iterator has value %s\n",
                   iter->key().ToString(true).c_str());
@@ -1681,6 +1703,37 @@ Status StressTest::TestBackupRestore(
   } else {
     backup_opts.schema_version = 2;
   }
+  if (thread->rand.OneIn(3)) {
+    backup_opts.max_background_operations = 16;
+  } else {
+    backup_opts.max_background_operations = 1;
+  }
+  if (thread->rand.OneIn(2)) {
+    backup_opts.backup_rate_limiter.reset(NewGenericRateLimiter(
+        FLAGS_backup_max_size * 1000000 /* rate_bytes_per_sec */,
+        1 /* refill_period_us */));
+  }
+  if (thread->rand.OneIn(2)) {
+    backup_opts.restore_rate_limiter.reset(NewGenericRateLimiter(
+        FLAGS_backup_max_size * 1000000 /* rate_bytes_per_sec */,
+        1 /* refill_period_us */));
+  }
+  std::ostringstream backup_opt_oss;
+  backup_opt_oss << "share_table_files: " << backup_opts.share_table_files
+                 << ", share_files_with_checksum: "
+                 << backup_opts.share_files_with_checksum
+                 << ", share_files_with_checksum_naming: "
+                 << backup_opts.share_files_with_checksum_naming
+                 << ", schema_version: " << backup_opts.schema_version
+                 << ", max_background_operations: "
+                 << backup_opts.max_background_operations
+                 << ", backup_rate_limiter: "
+                 << backup_opts.backup_rate_limiter.get()
+                 << ", restore_rate_limiter: "
+                 << backup_opts.restore_rate_limiter.get();
+
+  std::ostringstream create_backup_opt_oss;
+  std::ostringstream restore_opts_oss;
   BackupEngine* backup_engine = nullptr;
   std::string from = "a backup/restore operation";
   Status s = BackupEngine::Open(db_stress_env, backup_opts, &backup_engine);
@@ -1707,6 +1760,16 @@ Status StressTest::TestBackupRestore(
       // lock and wait on a background operation (flush).
       create_opts.flush_before_backup = true;
     }
+    create_opts.decrease_background_thread_cpu_priority = thread->rand.OneIn(2);
+    create_opts.background_thread_cpu_priority = static_cast<CpuPriority>(
+        thread->rand.Next() % (static_cast<int>(CpuPriority::kHigh) + 1));
+    create_backup_opt_oss << "flush_before_backup: "
+                          << create_opts.flush_before_backup
+                          << ", decrease_background_thread_cpu_priority: "
+                          << create_opts.decrease_background_thread_cpu_priority
+                          << ", background_thread_cpu_priority: "
+                          << static_cast<int>(
+                                 create_opts.background_thread_cpu_priority);
     s = backup_engine->CreateNewBackup(create_opts, db_);
     if (!s.ok()) {
       from = "BackupEngine::CreateNewBackup";
@@ -1744,19 +1807,21 @@ Status StressTest::TestBackupRestore(
   const bool allow_persistent = thread->tid == 0;  // not too many
   bool from_latest = false;
   int count = static_cast<int>(backup_info.size());
+  RestoreOptions restore_options;
+  restore_options.keep_log_files = thread->rand.OneIn(2);
+  restore_opts_oss << "keep_log_files: " << restore_options.keep_log_files;
   if (s.ok() && !inplace_not_restore) {
     if (count > 1) {
       s = backup_engine->RestoreDBFromBackup(
-          RestoreOptions(), backup_info[thread->rand.Uniform(count)].backup_id,
+          restore_options, backup_info[thread->rand.Uniform(count)].backup_id,
           restore_dir /* db_dir */, restore_dir /* wal_dir */);
       if (!s.ok()) {
         from = "BackupEngine::RestoreDBFromBackup";
       }
     } else {
       from_latest = true;
-      s = backup_engine->RestoreDBFromLatestBackup(RestoreOptions(),
-                                                   restore_dir /* db_dir */,
-                                                   restore_dir /* wal_dir */);
+      s = backup_engine->RestoreDBFromLatestBackup(
+          restore_options, restore_dir /* db_dir */, restore_dir /* wal_dir */);
       if (!s.ok()) {
         from = "BackupEngine::RestoreDBFromLatestBackup";
       }
@@ -1779,9 +1844,9 @@ Status StressTest::TestBackupRestore(
   std::vector<ColumnFamilyHandle*> restored_cf_handles;
 
   // Not yet implemented: opening restored BlobDB or TransactionDB
-  Options restore_options;
+  Options db_opt;
   if (s.ok() && !FLAGS_use_txn && !FLAGS_use_blob_db) {
-    s = PrepareOptionsForRestoredDB(&restore_options);
+    s = PrepareOptionsForRestoredDB(&db_opt);
     if (!s.ok()) {
       from = "PrepareRestoredDBOptions in backup/restore";
     }
@@ -1794,19 +1859,19 @@ Status StressTest::TestBackupRestore(
     // the same order as `column_family_names_`.
     assert(FLAGS_clear_column_family_one_in == 0);
     for (const auto& name : column_family_names_) {
-      cf_descriptors.emplace_back(name, ColumnFamilyOptions(restore_options));
+      cf_descriptors.emplace_back(name, ColumnFamilyOptions(db_opt));
     }
     if (inplace_not_restore) {
       BackupInfo& info = backup_info[thread->rand.Uniform(count)];
-      restore_options.env = info.env_for_open.get();
-      s = DB::OpenForReadOnly(DBOptions(restore_options), info.name_for_open,
+      db_opt.env = info.env_for_open.get();
+      s = DB::OpenForReadOnly(DBOptions(db_opt), info.name_for_open,
                               cf_descriptors, &restored_cf_handles,
                               &restored_db);
       if (!s.ok()) {
         from = "DB::OpenForReadOnly in backup/restore";
       }
     } else {
-      s = DB::Open(DBOptions(restore_options), restore_dir, cf_descriptors,
+      s = DB::Open(DBOptions(db_opt), restore_dir, cf_descriptors,
                    &restored_cf_handles, &restored_db);
       if (!s.ok()) {
         from = "DB::Open in backup/restore";
@@ -1898,8 +1963,13 @@ Status StressTest::TestBackupRestore(
     }
   }
   if (!s.ok() && (!s.IsIOError() || !std::strstr(s.getState(), "injected"))) {
-    fprintf(stderr, "Failure in %s with: %s\n", from.c_str(),
-            s.ToString().c_str());
+    fprintf(stderr,
+            "Failure in %s with: %s under specified BackupEngineOptions: %s, "
+            "CreateBackupOptions: %s, RestoreOptions: %s (Empty string or "
+            "missing field indicates default option or value is used)\n",
+            from.c_str(), s.ToString().c_str(), backup_opt_oss.str().c_str(),
+            create_backup_opt_oss.str().c_str(),
+            restore_opts_oss.str().c_str());
   }
   return s;
 }
@@ -2248,12 +2318,27 @@ void StressTest::TestCompactFiles(ThreadState* thread,
 
       size_t output_level =
           std::min(random_level + 1, cf_meta_data.levels.size() - 1);
-      auto s = db_->CompactFiles(CompactionOptions(), column_family,
-                                 input_files, static_cast<int>(output_level));
-      if (!s.ok()) {
-        fprintf(stdout, "Unable to perform CompactFiles(): %s\n",
-                s.ToString().c_str());
+      CompactionOptions compact_options;
+      if (thread->rand.OneIn(2)) {
+        compact_options.output_file_size_limit = FLAGS_target_file_size_base;
+      }
+      std::ostringstream compact_opt_oss;
+      compact_opt_oss << "output_file_size_limit: "
+                      << compact_options.output_file_size_limit;
+      auto s = db_->CompactFiles(compact_options, column_family, input_files,
+                                 static_cast<int>(output_level));
+      if (!s.ok() && !s.IsManualCompactionPaused() && !s.IsAborted() &&
+          !std::strstr(s.getState(), "injected") &&
+          !(s.IsInvalidArgument() &&
+            std::strstr(s.getState(), "Specified compaction input file") &&
+            std::strstr(s.getState(), "does not exist in column family"))) {
+        fprintf(stderr,
+                "Unable to perform CompactFiles(): %s under specified "
+                "CompactionOptions: %s (Empty string or "
+                "missing field indicates default option or value is used)\n",
+                s.ToString().c_str(), compact_opt_oss.str().c_str());
         thread->stats.AddNumCompactFilesFailed(1);
+        thread->shared->SafeTerminate();
       } else {
         thread->stats.AddNumCompactFilesSucceed(1);
       }
@@ -2396,6 +2481,9 @@ void StressTest::TestCompactRange(ThreadState* thread, int64_t rand_key,
   CompactRangeOptions cro;
   cro.exclusive_manual_compaction = static_cast<bool>(thread->rand.Next() % 2);
   cro.change_level = static_cast<bool>(thread->rand.Next() % 2);
+  if (thread->rand.OneIn(2)) {
+    cro.target_level = thread->rand.Next() % options_.num_levels;
+  }
   std::vector<BottommostLevelCompaction> bottom_level_styles = {
       BottommostLevelCompaction::kSkip,
       BottommostLevelCompaction::kIfHaveCompactionFilter,
@@ -2425,12 +2513,36 @@ void StressTest::TestCompactRange(ThreadState* thread, int64_t rand_key,
     pre_hash =
         GetRangeHash(thread, pre_snapshot, column_family, start_key, end_key);
   }
-
+  std::ostringstream compact_range_opt_oss;
+  compact_range_opt_oss << "exclusive_manual_compaction: "
+                        << cro.exclusive_manual_compaction
+                        << ", change_level: " << cro.change_level
+                        << ", target_level: " << cro.target_level
+                        << ", bottommost_level_compaction: "
+                        << static_cast<int>(cro.bottommost_level_compaction)
+                        << ", allow_write_stall: " << cro.allow_write_stall
+                        << ", max_subcompactions: " << cro.max_subcompactions
+                        << ", blob_garbage_collection_policy: "
+                        << static_cast<int>(cro.blob_garbage_collection_policy)
+                        << ", blob_garbage_collection_age_cutoff: "
+                        << cro.blob_garbage_collection_age_cutoff;
   Status status = db_->CompactRange(cro, column_family, &start_key, &end_key);
 
-  if (!status.ok()) {
-    fprintf(stdout, "Unable to perform CompactRange(): %s\n",
-            status.ToString().c_str());
+  if (!status.ok() && !status.IsManualCompactionPaused() &&
+      !std::strstr(status.getState(), "injected") &&
+      !(status.IsNotSupported() &&
+        (std::strstr(status.getState(), "another thread is refitting") ||
+         std::strstr(
+             status.getState(),
+             "Levels between source and target are not empty for a move")))) {
+    fprintf(stdout,
+            "Unable to perform CompactRange(): %s under specified "
+            "CompactRangeOptions: %s (Empty string or "
+            "missing field indicates default option or value is used)\n",
+            status.ToString().c_str(), compact_range_opt_oss.str().c_str());
+    thread->stats.AddErrors(1);
+    // Fail fast to preserve the DB state.
+    thread->shared->SetVerificationFailure();
   }
 
   if (pre_snapshot != nullptr) {
@@ -2439,8 +2551,11 @@ void StressTest::TestCompactRange(ThreadState* thread, int64_t rand_key,
     if (pre_hash != post_hash) {
       fprintf(stderr,
               "Data hash different before and after compact range "
-              "start_key %s end_key %s\n",
-              start_key.ToString(true).c_str(), end_key.ToString(true).c_str());
+              "start_key %s end_key %s under specified CompactRangeOptions: %s "
+              "(Empty string or "
+              "missing field indicates default option or value is used)\n",
+              start_key.ToString(true).c_str(), end_key.ToString(true).c_str(),
+              compact_range_opt_oss.str().c_str());
       thread->stats.AddErrors(1);
       // Fail fast to preserve the DB state.
       thread->shared->SetVerificationFailure();
@@ -3286,6 +3401,15 @@ void InitializeOptionsFromFlags(
   block_based_options.max_auto_readahead_size = FLAGS_max_auto_readahead_size;
   block_based_options.num_file_reads_for_auto_readahead =
       FLAGS_num_file_reads_for_auto_readahead;
+  block_based_options.cache_index_and_filter_blocks_with_high_priority =
+      FLAGS_cache_index_and_filter_blocks_with_high_priority;
+  block_based_options.use_delta_encoding = FLAGS_use_delta_encoding;
+  block_based_options.verify_compression = FLAGS_verify_compression;
+  block_based_options.read_amp_bytes_per_bit = FLAGS_read_amp_bytes_per_bit;
+  block_based_options.enable_index_compression = FLAGS_enable_index_compression;
+  block_based_options.index_shortening =
+      static_cast<BlockBasedTableOptions::IndexShorteningMode>(
+          FLAGS_index_shortening);
   options.table_factory.reset(NewBlockBasedTableFactory(block_based_options));
   options.db_write_buffer_size = FLAGS_db_write_buffer_size;
   options.write_buffer_size = FLAGS_write_buffer_size;
@@ -3486,6 +3610,41 @@ void InitializeOptionsFromFlags(
 
   options.bottommost_file_compaction_delay =
       FLAGS_bottommost_file_compaction_delay;
+
+  options.allow_fallocate = FLAGS_allow_fallocate;
+  options.table_cache_numshardbits = FLAGS_table_cache_numshardbits;
+  options.log_readahead_size = FLAGS_log_readahead_size;
+  options.bgerror_resume_retry_interval = FLAGS_bgerror_resume_retry_interval;
+  options.delete_obsolete_files_period_micros =
+      FLAGS_delete_obsolete_files_period_micros;
+  options.max_log_file_size = FLAGS_max_log_file_size;
+  options.log_file_time_to_roll = FLAGS_log_file_time_to_roll;
+  options.use_adaptive_mutex = FLAGS_use_adaptive_mutex;
+  options.advise_random_on_open = FLAGS_advise_random_on_open;
+  // TODO (hx235): test the functionality of `WAL_ttl_seconds`,
+  // `WAL_size_limit_MB` i.e, `GetUpdatesSince()`
+  options.WAL_ttl_seconds = FLAGS_WAL_ttl_seconds;
+  options.WAL_size_limit_MB = FLAGS_WAL_size_limit_MB;
+  options.wal_bytes_per_sync = FLAGS_wal_bytes_per_sync;
+  options.strict_bytes_per_sync = FLAGS_strict_bytes_per_sync;
+  options.avoid_flush_during_shutdown = FLAGS_avoid_flush_during_shutdown;
+  options.dump_malloc_stats = FLAGS_dump_malloc_stats;
+  options.stats_history_buffer_size = FLAGS_stats_history_buffer_size;
+  options.skip_stats_update_on_db_open = FLAGS_skip_stats_update_on_db_open;
+  options.optimize_filters_for_hits = FLAGS_optimize_filters_for_hits;
+  options.sample_for_compression = FLAGS_sample_for_compression;
+  options.report_bg_io_stats = FLAGS_report_bg_io_stats;
+  options.manifest_preallocation_size = FLAGS_manifest_preallocation_size;
+  if (FLAGS_enable_checksum_handoff) {
+    options.checksum_handoff_file_types = {FileTypeSet::All()};
+  } else {
+    options.checksum_handoff_file_types = {};
+  }
+  options.max_total_wal_size = FLAGS_max_total_wal_size;
+  options.soft_pending_compaction_bytes_limit =
+      FLAGS_soft_pending_compaction_bytes_limit;
+  options.hard_pending_compaction_bytes_limit =
+      FLAGS_hard_pending_compaction_bytes_limit;
 }
 
 void InitializeOptionsGeneral(
diff --git a/db_stress_tool/no_batched_ops_stress.cc b/db_stress_tool/no_batched_ops_stress.cc
index 7b0c6d2eb..a94de3897 100644
--- a/db_stress_tool/no_batched_ops_stress.cc
+++ b/db_stress_tool/no_batched_ops_stress.cc
@@ -1526,6 +1526,7 @@ class NonBatchedOpsStressTest : public StressTest {
     const std::string sst_filename =
         FLAGS_db + "/." + std::to_string(thread->tid) + ".sst";
     Status s;
+    std::ostringstream ingest_options_oss;
     if (db_stress_env->FileExists(sst_filename).ok()) {
       // Maybe we terminated abnormally before, so cleanup to give this file
       // ingestion a clean slate
@@ -1594,8 +1595,18 @@ class NonBatchedOpsStressTest : public StressTest {
       s = sst_file_writer.Finish();
     }
     if (s.ok()) {
+      IngestExternalFileOptions ingest_options;
+      ingest_options.move_files = thread->rand.OneInOpt(2);
+      ingest_options.verify_checksums_before_ingest = thread->rand.OneInOpt(2);
+      ingest_options.verify_checksums_readahead_size =
+          thread->rand.OneInOpt(2) ? 1024 * 1024 : 0;
+      ingest_options_oss << "move_files: " << ingest_options.move_files
+                         << ", verify_checksums_before_ingest: "
+                         << ingest_options.verify_checksums_before_ingest
+                         << ", verify_checksums_readahead_size: "
+                         << ingest_options.verify_checksums_readahead_size;
       s = db_->IngestExternalFile(column_families_[column_family],
-                                  {sst_filename}, IngestExternalFileOptions());
+                                  {sst_filename}, ingest_options);
     }
     if (!s.ok()) {
       for (PendingExpectedValue& pending_expected_value :
@@ -1603,7 +1614,11 @@ class NonBatchedOpsStressTest : public StressTest {
         pending_expected_value.Rollback();
       }
       if (!s.IsIOError() || !std::strstr(s.getState(), "injected")) {
-        fprintf(stderr, "file ingestion error: %s\n", s.ToString().c_str());
+        fprintf(stderr,
+                "file ingestion error: %s under specified "
+                "IngestExternalFileOptions: %s (Empty string or "
+                "missing field indicates default option or value is used)\n",
+                s.ToString().c_str(), ingest_options_oss.str().c_str());
         thread->shared->SafeTerminate();
       } else {
         fprintf(stdout, "file ingestion error: %s\n", s.ToString().c_str());
diff --git a/tools/db_crashtest.py b/tools/db_crashtest.py
index 7037156b5..ddd1b9eca 100644
--- a/tools/db_crashtest.py
+++ b/tools/db_crashtest.py
@@ -227,8 +227,47 @@ default_params = {
     ),
     "auto_readahead_size" : lambda: random.choice([0, 1]),
     "verify_iterator_with_expected_state_one_in": 5,
+    "allow_fallocate": lambda: random.choice([0, 1]),
+    "table_cache_numshardbits": lambda: random.choice([6] * 3 + [-1] * 2 + [0]),
+    "enable_write_thread_adaptive_yield": lambda: random.choice([0, 1]),
+    "log_readahead_size": lambda: random.choice([0, 16 * 1024 * 1024]),
+    "bgerror_resume_retry_interval": lambda: random.choice([10000, 1000000]),
+    "delete_obsolete_files_period_micros": lambda: random.choice([6 * 60 * 60 * 1000000, 30 * 1000000]),
+    "max_log_file_size": lambda: random.choice([0, 1024 * 1024]),
+    "log_file_time_to_roll": lambda: random.choice([0, 60]),
+    "use_adaptive_mutex": lambda: random.choice([0, 1]),
+    "advise_random_on_open": lambda: random.choice([0] + [1] * 3),
+    "WAL_ttl_seconds": lambda: random.choice([0, 60]),
+    "WAL_size_limit_MB": lambda: random.choice([0, 1]),
+    "wal_bytes_per_sync": lambda: random.choice([0, 1024 * 1024]),
+    "strict_bytes_per_sync": lambda: random.choice([0, 1]),
+    "avoid_flush_during_shutdown": lambda: random.choice([0, 1]),
+    "fill_cache": lambda: random.choice([0, 1]),
+    "optimize_multiget_for_io": lambda: random.choice([0, 1]),
+    "memtable_insert_hint_per_batch": lambda: random.choice([0, 1]),
+    "dump_malloc_stats": lambda: random.choice([0, 1]),
+    "stats_history_buffer_size": lambda: random.choice([0, 1024 * 1024]),
+    "skip_stats_update_on_db_open": lambda: random.choice([0, 1]),
+    "optimize_filters_for_hits": lambda: random.choice([0, 1]),
+    "sample_for_compression": lambda: random.choice([0, 5]),
+    "report_bg_io_stats": lambda: random.choice([0, 1]),
+    "cache_index_and_filter_blocks_with_high_priority": lambda: random.choice([0, 1]),
+    "use_delta_encoding": lambda: random.choice([0, 1]),
+    "verify_compression": lambda: random.choice([0, 1]),
+    "read_amp_bytes_per_bit": lambda: random.choice([0, 32]),
+    "enable_index_compression": lambda: random.choice([0, 1]),
+    "index_shortening": lambda: random.choice([0, 1, 2]),
+    "metadata_charge_policy": lambda: random.choice([0, 1]),
+    "use_adaptive_mutex_lru": lambda: random.choice([0, 1]),
+    "compress_format_version": lambda: random.choice([1, 2]),
+    "manifest_preallocation_size": lambda: random.choice([0, 5 * 1024]),
+    "enable_checksum_handoff": lambda: random.choice([0, 1]),
+    "max_total_wal_size": lambda: random.choice([0] * 4 + [64 * 1024 * 1024]),
+    "high_pri_pool_ratio": lambda: random.choice([0, 0.5]),
+    "low_pri_pool_ratio": lambda: random.choice([0, 0.5]),
+    "soft_pending_compaction_bytes_limit" : lambda: random.choice([1024 * 1024] + [64 * 1073741824] * 4),
+    "hard_pending_compaction_bytes_limit" : lambda: random.choice([2 * 1024 * 1024] + [256 * 1073741824] * 4),
 }
-
 _TEST_DIR_ENV_VAR = "TEST_TMPDIR"
 # If TEST_TMPDIR_EXPECTED is not specified, default value will be TEST_TMPDIR
 _TEST_EXPECTED_DIR_ENV_VAR = "TEST_TMPDIR_EXPECTED"