Pass statistics and fix null clock in SstFileReader::MultiGet (#14393)

Summary:
**Summary:**
This is to sync an internal change of passing `Statistics*` to the GetContext constructor and collecting more stats as well as fix a bug this change created.

SstFileReader::MultiGet was passing nullptr for both `SystemClock*` and `Statistics*` to the GetContext constructor.  After  `Statistics*` was passed to the GetContext constructor (the internal change), this caused a segfault when a merge operation was triggered with statistics enabled, because the merge helper's StopWatchNano attempted to dereference the null clock pointer. Fix by passing `r->ioptions.clock` from the reader's options.

Additionally, add `assert(clock_)` guards to `StopWatchNano::Start()` and `ElapsedNanos()` to catch null clock bugs in debug builds. Can't do so in release build because it's on hot path.

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

Test Plan:
- `./sst_file_reader_test --gtest_filter='SstFileReaderTableMultiGetTest.Basic'` exercises merge with statistics enabled, previously segfaulted without the fix with the clock, now passes.
- `./sst_file_reader_test` - all tests pass.

Reviewed By: xingbowang

Differential Revision: D94599343

Pulled By: hx235

fbshipit-source-id: 0a748bb00ee27bb202d01d410b52657101c05de0

table/sst_file_reader.cc

@@ -93,6 +93,7 @@ std::vector<Status> SstFileReader::MultiGet(const ReadOptions& roptions,
   auto r = rep_.get();
   const Comparator* user_comparator =
       r->ioptions.internal_comparator.user_comparator();
+  Statistics* statistics = r->ioptions.stats;
 
   autovector<KeyContext, MultiGetContext::MAX_BATCH_SIZE> key_context;
   autovector<KeyContext*, MultiGetContext::MAX_BATCH_SIZE> sorted_keys;
@@ -105,11 +106,12 @@ std::vector<Status> SstFileReader::MultiGet(const ReadOptions& roptions,
     merge_ctx.emplace_back();
     key_context.emplace_back(nullptr, keys[i], val, nullptr,
                              nullptr /* timestamp */, &statuses[i]);
-    get_ctx.emplace_back(user_comparator, r->ioptions.merge_operator.get(),
-                         nullptr, nullptr, GetContext::kNotFound,
-                         *key_context[i].key, val, nullptr, nullptr, nullptr,
-                         &merge_ctx[i], true,
-                         &key_context[i].max_covering_tombstone_seq, nullptr);
+    get_ctx.emplace_back(
+        user_comparator, r->ioptions.merge_operator.get(), nullptr /* logger */,
+        statistics, GetContext::kNotFound, *key_context[i].key, val,
+        nullptr /* columns */, nullptr /* timestamp */,
+        nullptr /* value_found */, &merge_ctx[i], true,
+        &key_context[i].max_covering_tombstone_seq, r->ioptions.clock);
     key_context[i].get_context = &get_ctx[i];
   }
   for (size_t i = 0; i < num_keys; ++i) {
@@ -139,6 +141,8 @@ std::vector<Status> SstFileReader::MultiGet(const ReadOptions& roptions,
 
   values->resize(num_keys);
   for (size_t i = 0; i < num_keys; ++i) {
+    get_ctx[i].ReportCounters();
+
     if (statuses[i].ok()) {
       switch (get_ctx[i].State()) {
         case GetContext::kFound:

table/sst_file_reader_test.cc

@@ -801,6 +801,7 @@ TEST_F(SstFileReaderTableMultiGetTest, Basic) {
   options.comparator = ucmp;
   options.disable_auto_compactions = true;
   options.merge_operator = MergeOperators::CreateStringAppendOperator();
+  options.statistics = CreateDBStatistics();
   BlockBasedTableOptions bbto;
   bbto.filter_policy.reset(NewBloomFilterPolicy(10, false));
   options.table_factory.reset(NewBlockBasedTableFactory(bbto));
@@ -830,6 +831,8 @@ TEST_F(SstFileReaderTableMultiGetTest, Basic) {
   ASSERT_OK(reader.Open(file_name));
   ASSERT_OK(reader.VerifyChecksum());
 
+  ASSERT_OK(options.statistics->Reset());
+
   std::vector<Slice> keys;
   std::vector<std::string> values;
 
@@ -850,6 +853,10 @@ TEST_F(SstFileReaderTableMultiGetTest, Basic) {
   ASSERT_TRUE(statuses[4].ok());
   ASSERT_EQ("val4,val5", values[4]);
 
+  uint64_t cache_hits = options.statistics->getTickerCount(BLOCK_CACHE_HIT);
+  uint64_t cache_misses = options.statistics->getTickerCount(BLOCK_CACHE_MISS);
+  ASSERT_GT(cache_hits + cache_misses, 0);
+
   dbfull()->ReleaseSnapshot(snapshot1);
   dbfull()->ReleaseSnapshot(snapshot2);
   dbfull()->ReleaseSnapshot(snapshot3);

util/stop_watch.h

@@ -112,6 +112,7 @@ class StopWatchNano {
     }
   }
   void Start() {
+    assert(clock_);
     if constexpr (use_cpu_time) {
       start_ = clock_->CPUNanos();
     } else {
@@ -119,6 +120,7 @@ class StopWatchNano {
     }
   }
   uint64_t ElapsedNanos(bool reset = false) {
+    assert(clock_);
     uint64_t now = 0;
     if constexpr (use_cpu_time) {
       now = clock_->CPUNanos();