Fix improper ExpectedValute::Exists() usages and disable compaction during VerifyDB() in crash test (#12933)

Summary:
**Context:**
Adding assertion `!PendingPut()&&!PendingDelete()` in `ExpectedValute::Exists()` surfaced a couple improper usages of `ExpectedValute::Exists()` in the crash test
- Commit phase of `ExpectedValue::Delete()`/`SyncDelete()`:
When we issue delete to expected value during commit phase or `SyncDelete()` (used in crash recovery verification) as below, we don't really care about the result.
d458331ee9/db_stress_tool/expected_state.cc (L73)
d458331ee9/db_stress_tool/expected_value.cc (L52)
That means, we don't really need to check for `Exists()` d458331ee9/db_stress_tool/expected_value.cc (L24-L26).
This actually gives an alternative solution to b65e29a4a9 to solve false-positive assertion violation.
- TestMultiGetXX() path: `Exists()` is called without holding the lock as required

f63428bcc7/db_stress_tool/no_batched_ops_stress.cc (L2688)
```
void MaybeAddKeyToTxnForRYW(
      ThreadState* thread, int column_family, int64_t key, Transaction* txn,
      std::unordered_map<std::string, ExpectedValue>& ryw_expected_values) {
    assert(thread);
    assert(txn);

    SharedState* const shared = thread->shared;
    assert(shared);

    if (!shared->AllowsOverwrite(key) && shared->Exists(column_family, key)) {
      // Just do read your write checks for keys that allow overwrites.
      return;
    }

    // With a 1 in 10 probability, insert the just added key in the batch
    // into the transaction. This will create an overlap with the MultiGet
    // keys and exercise some corner cases in the code
    if (thread->rand.OneIn(10)) {
```

f63428bcc7/db_stress_tool/expected_state.h (L74-L76)

The assertion also failed if db stress compaction filter was invoked before crash recovery verification (`VerifyDB()`->`VerifyOrSyncValue()`) finishes.
f63428bcc7/db_stress_tool/db_stress_compaction_filter.h (L53)
It failed because it can encounter a key with pending state when checking for `Exists()` since that key's expected state has not been sync-ed with db state in `VerifyOrSyncValue()`.
f63428bcc7/db_stress_tool/no_batched_ops_stress.cc (L2579-L2591)

**Summary:**
This PR fixes above issues by
- not checking `Exists()` in commit phase/`SyncDelete()`
- using the concurrent version of key existence check like in other read
- conditionally temporarily disabling compaction till after crash recovery verification succeeds()

And add back the assertion `!PendingPut()&&!PendingDelete()`

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

Test Plan: Rehearsal CI

Reviewed By: cbi42

Differential Revision: D61214889

Pulled By: hx235

fbshipit-source-id: ef25ba896e64330ddf330182314981516880c3e4

db_stress_tool/db_stress_common.h

@@ -125,7 +125,6 @@ DECLARE_int32(level0_stop_writes_trigger);
 DECLARE_int32(block_size);
 DECLARE_int32(format_version);
 DECLARE_int32(index_block_restart_interval);
-DECLARE_bool(disable_auto_compactions);
 DECLARE_int32(max_background_compactions);
 DECLARE_int32(num_bottom_pri_threads);
 DECLARE_int32(compaction_thread_pool_adjust_interval);
@@ -318,7 +317,6 @@ DECLARE_int32(prepopulate_blob_cache);
 DECLARE_int32(approximate_size_one_in);
 DECLARE_bool(best_efforts_recovery);
 DECLARE_bool(skip_verifydb);
-DECLARE_bool(enable_compaction_filter);
 DECLARE_bool(paranoid_file_checks);
 DECLARE_bool(fail_if_options_file_error);
 DECLARE_uint64(batch_protection_bytes_per_key);

db_stress_tool/db_stress_compaction_filter.h

@@ -49,7 +49,7 @@ class DbStressCompactionFilter : public CompactionFilter {
       return Decision::kKeep;
     }
     // Reaching here means we acquired the lock.
-
+    key_mutex->AssertHeld();
     bool key_exists = state_->Exists(cf_id_, key_num);
     const bool allow_overwrite = state_->AllowsOverwrite(key_num);
 

db_stress_tool/db_stress_driver.cc

@@ -167,7 +167,10 @@ bool RunStressTestImpl(SharedState* shared) {
               {FileType::kWalFile});
         }
       }
-      now = clock->NowMicros();
+      if (ShouldDisableAutoCompactionsBeforeVerifyDb()) {
+        Status s = stress->EnableAutoCompaction();
+        assert(s.ok());
+      }
       fprintf(stdout, "%s Starting database operations\n",
               clock->TimeToString(now / 1000000).c_str());
 

db_stress_tool/db_stress_shared_state.h

@@ -45,6 +45,8 @@ DECLARE_int32(open_write_fault_one_in);
 DECLARE_int32(open_read_fault_one_in);
 
 DECLARE_int32(inject_error_severity);
+DECLARE_bool(disable_auto_compactions);
+DECLARE_bool(enable_compaction_filter);
 
 namespace ROCKSDB_NAMESPACE {
 class StressTest;

db_stress_tool/db_stress_test_base.cc

@@ -3832,6 +3832,10 @@ void CheckAndSetOptionsForUserTimestamp(Options& options) {
       FLAGS_persist_user_defined_timestamps;
 }
 
+bool ShouldDisableAutoCompactionsBeforeVerifyDb() {
+  return !FLAGS_disable_auto_compactions && FLAGS_enable_compaction_filter;
+}
+
 bool InitializeOptionsFromFile(Options& options) {
   DBOptions db_options;
   ConfigOptions config_options;
@@ -3945,7 +3949,11 @@ void InitializeOptionsFromFlags(
         new WriteBufferManager(FLAGS_db_write_buffer_size, block_cache));
   }
   options.memtable_whole_key_filtering = FLAGS_memtable_whole_key_filtering;
-  options.disable_auto_compactions = FLAGS_disable_auto_compactions;
+  if (ShouldDisableAutoCompactionsBeforeVerifyDb()) {
+    options.disable_auto_compactions = true;
+  } else {
+    options.disable_auto_compactions = FLAGS_disable_auto_compactions;
+  }
   options.max_background_compactions = FLAGS_max_background_compactions;
   options.max_background_flushes = FLAGS_max_background_flushes;
   options.compaction_style =

db_stress_tool/db_stress_test_base.h

@@ -48,7 +48,11 @@ class StressTest {
     return FLAGS_sync_fault_injection || FLAGS_disable_wal ||
            FLAGS_manual_wal_flush_one_in > 0;
   }
-
+  Status EnableAutoCompaction() {
+    assert(options_.disable_auto_compactions);
+    Status s = db_->EnableAutoCompaction(column_families_);
+    return s;
+  }
   void CleanUp();
 
  protected:
@@ -447,5 +451,6 @@ void InitializeOptionsGeneral(
 // user-defined timestamp.
 void CheckAndSetOptionsForUserTimestamp(Options& options);
 
+bool ShouldDisableAutoCompactionsBeforeVerifyDb();
 }  // namespace ROCKSDB_NAMESPACE
 #endif  // GFLAGS

db_stress_tool/expected_value.cc

@@ -21,7 +21,7 @@ void ExpectedValue::Put(bool pending) {
 }
 
 bool ExpectedValue::Delete(bool pending) {
-  if (!Exists()) {
+  if (pending && !Exists()) {
     return false;
   }
   if (pending) {

db_stress_tool/expected_value.h

@@ -38,7 +38,7 @@ class ExpectedValue {
       : expected_value_(expected_value) {}
 
   bool Exists() const {
-    assert(!PendingWrite());
+    assert(!PendingWrite() && !PendingDelete());
     return !IsDeleted();
   }
 

db_stress_tool/no_batched_ops_stress.cc

@@ -2587,6 +2587,8 @@ class NonBatchedOpsStressTest : public StressTest {
         // Value doesn't exist in db, update state to reflect that
         shared->SyncDelete(cf, key);
         return true;
+      } else {
+        assert(false);
       }
     }
     char expected_value_data[kValueMaxLen];
@@ -2685,7 +2687,11 @@ class NonBatchedOpsStressTest : public StressTest {
     SharedState* const shared = thread->shared;
     assert(shared);
 
-    if (!shared->AllowsOverwrite(key) && shared->Exists(column_family, key)) {
+    const ExpectedValue expected_value =
+        thread->shared->Get(column_family, key);
+    bool may_exist = !ExpectedValueHelper::MustHaveNotExisted(expected_value,
+                                                              expected_value);
+    if (!shared->AllowsOverwrite(key) && may_exist) {
       // Just do read your write checks for keys that allow overwrites.
       return;
     }