Add allow_unprepared_value+PrepareValue() to the stress tests (#13125)

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

The patch adds the new read option `allow_unprepared_value` and the new `Iterator` / `CoalescingIterator` / `AttributeGroupIterator` API `PrepareValue()` to the stress/crash tests. The change affects the batched, non-batched, and CF consistency stress test flavors and the `TestIterate`, `TestPrefixScan`, and `TestIterateAgainstExpected` operations.

Reviewed By: hx235

Differential Revision: D65636380

fbshipit-source-id: fd0caa0e87d03b6206667f07499b0c11847d1bbe

db_stress_tool/batched_ops_stress.cc

@@ -613,6 +613,15 @@ class BatchedOpsStressTest : public StressTest {
         // no iterator should finish before the first one
         assert(iters[i]->Valid() &&
                iters[i]->key().starts_with(prefix_slices[i]));
+
+        if (ro_copies[i].allow_unprepared_value) {
+          if (!iters[i]->PrepareValue()) {
+            fprintf(stderr,
+                    "prefix scan error: PrepareValue failed for key %s\n",
+                    iters[i]->key().ToString(/* hex */ true).c_str());
+          }
+        }
+
         values[i] = iters[i]->value().ToString();
 
         // make sure the last character of the value is the expected digit

db_stress_tool/cf_consistency_stress.cc

@@ -802,6 +802,13 @@ class CfConsistencyStressTest : public StressTest {
          iter->Next()) {
       ++count;
 
+      if (ro_copy.allow_unprepared_value) {
+        if (!iter->PrepareValue()) {
+          s = iter->status();
+          break;
+        }
+      }
+
       if (!VerifyWideColumns(iter->value(), iter->columns())) {
         s = Status::Corruption("Value and columns inconsistent",
                                DebugString(iter->value(), iter->columns()));

db_stress_tool/db_stress_common.h

@@ -418,6 +418,7 @@ DECLARE_bool(check_multiget_entity_consistency);
 DECLARE_bool(inplace_update_support);
 DECLARE_uint32(uncache_aggressiveness);
 DECLARE_int32(test_ingest_standalone_range_deletion_one_in);
+DECLARE_bool(allow_unprepared_value);
 
 constexpr long KB = 1024;
 constexpr int kRandomValueMaxFactor = 3;

db_stress_tool/db_stress_gflags.cc

@@ -839,6 +839,10 @@ DEFINE_int32(test_ingest_standalone_range_deletion_one_in, 0,
              "If non-zero, file ingestion flow will test standalone range "
              "deletion file once every N file ingestion operations.");
 
+DEFINE_bool(allow_unprepared_value,
+            ROCKSDB_NAMESPACE::ReadOptions().allow_unprepared_value,
+            "Allow lazy loading of values for range scans");
+
 static bool ValidateInt32Percent(const char* flagname, int32_t value) {
   if (value < 0 || value > 100) {
     fprintf(stderr, "Invalid value for --%s: %d, 0<= pct <=100 \n", flagname,

db_stress_tool/db_stress_test_base.cc

@@ -918,6 +918,8 @@ void StressTest::OperateDb(ThreadState* thread) {
   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;
+  read_opts.allow_unprepared_value = FLAGS_allow_unprepared_value;
+
   WriteOptions write_opts;
   if (FLAGS_rate_limit_auto_wal_flush) {
     write_opts.rate_limiter_priority = Env::IO_USER;
@@ -1906,7 +1908,9 @@ void StressTest::VerifyIterator(
                << ", iterate_lower_bound: "
                << (ro.iterate_lower_bound
                        ? ro.iterate_lower_bound->ToString(true).c_str()
-                       : "");
+                       : "")
+               << ", allow_unprepared_value: " << ro.allow_unprepared_value;
+
   if (iter->Valid() && !cmp_iter->Valid()) {
     if (pe != nullptr) {
       if (!pe->InDomain(seek_key)) {
@@ -1996,10 +2000,26 @@ void StressTest::VerifyIterator(
   }
 
   if (!*diverged && iter->Valid()) {
-    if (!verify_func(iter)) {
-      *diverged = true;
+    if (ro.allow_unprepared_value) {
+      if (!iter->PrepareValue()) {
+        fprintf(
+            stderr,
+            "Iterator failed to prepare value for 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(), op_logs.c_str(),
+            read_opt_oss.str().c_str());
+        *diverged = true;
+      }
+    }
+
+    if (!*diverged && iter->Valid()) {
+      if (!verify_func(iter)) {
+        *diverged = true;
+      }
     }
   }
+
   if (*diverged) {
     fprintf(stderr, "VerifyIterator failed. Control CF %s\n",
             cmp_cfh->GetName().c_str());

db_stress_tool/no_batched_ops_stress.cc

@@ -1515,6 +1515,13 @@ class NonBatchedOpsStressTest : public StressTest {
         }
       }
 
+      if (ro_copy.allow_unprepared_value) {
+        if (!iter->PrepareValue()) {
+          s = iter->status();
+          break;
+        }
+      }
+
       if (!VerifyWideColumns(iter->value(), iter->columns())) {
         s = Status::Corruption("Value and columns inconsistent",
                                DebugString(iter->value(), iter->columns()));
@@ -2293,6 +2300,22 @@ class NonBatchedOpsStressTest : public StressTest {
       assert(iter);
       assert(iter->Valid());
 
+      if (ro.allow_unprepared_value) {
+        if (!iter->PrepareValue()) {
+          shared->SetVerificationFailure();
+
+          fprintf(stderr,
+                  "Verification failed for key %s: failed to prepare value\n",
+                  Slice(iter->key()).ToString(/* hex */ true).c_str());
+          fprintf(stderr, "Column family: %s, op_logs: %s\n",
+                  cfh->GetName().c_str(), op_logs.c_str());
+
+          thread->stats.AddErrors(1);
+
+          return false;
+        }
+      }
+
       if (!VerifyWideColumns(iter->value(), iter->columns())) {
         shared->SetVerificationFailure();
 
@@ -2878,4 +2901,4 @@ StressTest* CreateNonBatchedOpsStressTest() {
 }
 
 }  // namespace ROCKSDB_NAMESPACE
-#endif  // GFLAGS
+#endif  // GFLAGS

tools/db_crashtest.py

@@ -342,6 +342,7 @@ default_params = {
     "use_timed_put_one_in": lambda: random.choice([0] * 7 + [1, 5, 10]),
     "universal_max_read_amp": lambda: random.choice([-1] * 3 + [0, 4, 10]),
     "paranoid_memory_checks": lambda: random.choice([0] * 7 + [1]),
+    "allow_unprepared_value": lambda: random.choice([0, 1]),
 }
 _TEST_DIR_ENV_VAR = "TEST_TMPDIR"
 # If TEST_TMPDIR_EXPECTED is not specified, default value will be TEST_TMPDIR