Fix perf_context::merge_operator_time_nanos calculation

Summary: We were not measuring the time spent in merge_operator when called from Version::Get()

Test Plan: added a unittest

Reviewers: sdong, yhchiang

Reviewed By: yhchiang

Subscribers: andrewkr, dhruba

Differential Revision: https://reviews.facebook.net/D55905

db/perf_context_test.cc

@@ -9,16 +9,16 @@
 #include <vector>
 
 #include "rocksdb/db.h"
+#include "rocksdb/memtablerep.h"
 #include "rocksdb/perf_context.h"
 #include "rocksdb/slice_transform.h"
-#include "rocksdb/memtablerep.h"
 #include "util/histogram.h"
 #include "util/instrumented_mutex.h"
 #include "util/stop_watch.h"
+#include "util/string_util.h"
 #include "util/testharness.h"
 #include "util/thread_status_util.h"
-#include "util/string_util.h"
+#include "utilities/merge_operators.h"
-
 
 bool FLAGS_random_key = false;
 bool FLAGS_use_set_based_memetable = false;
@@ -602,6 +602,41 @@ TEST_F(PerfContextTest, ToString) {
   ASSERT_EQ(std::string::npos, zero_excluded.find("= 0"));
   ASSERT_NE(std::string::npos, zero_excluded.find("= 12345"));
 }
+
+TEST_F(PerfContextTest, MergeOperatorTime) {
+  DestroyDB(kDbName, Options());
+  DB* db;
+  Options options;
+  options.create_if_missing = true;
+  options.merge_operator = MergeOperators::CreateStringAppendOperator();
+  Status s = DB::Open(options, kDbName, &db);
+  EXPECT_OK(s);
+
+  std::string val;
+  ASSERT_OK(db->Merge(WriteOptions(), "k1", "val1"));
+  ASSERT_OK(db->Merge(WriteOptions(), "k1", "val2"));
+  ASSERT_OK(db->Merge(WriteOptions(), "k1", "val3"));
+  ASSERT_OK(db->Merge(WriteOptions(), "k1", "val4"));
+
+  SetPerfLevel(kEnableTime);
+  perf_context.Reset();
+  ASSERT_OK(db->Get(ReadOptions(), "k1", &val));
+  EXPECT_GT(perf_context.merge_operator_time_nanos, 0);
+
+  ASSERT_OK(db->Flush(FlushOptions()));
+
+  perf_context.Reset();
+  ASSERT_OK(db->Get(ReadOptions(), "k1", &val));
+  EXPECT_GT(perf_context.merge_operator_time_nanos, 0);
+
+  ASSERT_OK(db->CompactRange(CompactRangeOptions(), nullptr, nullptr));
+
+  perf_context.Reset();
+  ASSERT_OK(db->Get(ReadOptions(), "k1", &val));
+  EXPECT_GT(perf_context.merge_operator_time_nanos, 0);
+
+  delete db;
+}
 }
 
 int main(int argc, char** argv) {

db/version_set.cc

@@ -23,6 +23,7 @@
 #include <vector>
 #include <string>
 
+#include "db/compaction.h"
 #include "db/filename.h"
 #include "db/internal_stats.h"
 #include "db/log_reader.h"
@@ -30,22 +31,22 @@
 #include "db/memtable.h"
 #include "db/merge_context.h"
 #include "db/table_cache.h"
-#include "db/compaction.h"
 #include "db/version_builder.h"
 #include "db/writebuffer.h"
 #include "rocksdb/env.h"
 #include "rocksdb/merge_operator.h"
-#include "table/internal_iterator.h"
-#include "table/table_reader.h"
-#include "table/merger.h"
-#include "table/two_level_iterator.h"
 #include "table/format.h"
-#include "table/plain_table_factory.h"
-#include "table/meta_blocks.h"
 #include "table/get_context.h"
+#include "table/internal_iterator.h"
+#include "table/merger.h"
+#include "table/meta_blocks.h"
+#include "table/plain_table_factory.h"
+#include "table/table_reader.h"
+#include "table/two_level_iterator.h"
 #include "util/coding.h"
 #include "util/file_reader_writer.h"
 #include "util/logging.h"
+#include "util/perf_context_imp.h"
 #include "util/stop_watch.h"
 #include "util/sync_point.h"
 
@@ -948,9 +949,16 @@ void Version::Get(const ReadOptions& read_options, const LookupKey& k,
     }
     // merge_operands are in saver and we hit the beginning of the key history
     // do a final merge of nullptr and operands;
-    if (merge_operator_->FullMerge(user_key, nullptr,
+    bool merge_success = false;
-                                   merge_context->GetOperands(), value,
+    {
-                                   info_log_)) {
+      StopWatchNano timer(env_, db_statistics_ != nullptr);
+      PERF_TIMER_GUARD(merge_operator_time_nanos);
+      merge_success = merge_operator_->FullMerge(
+          user_key, nullptr, merge_context->GetOperands(), value, info_log_);
+      RecordTick(db_statistics_, MERGE_OPERATION_TOTAL_TIME,
+                 timer.ElapsedNanos());
+    }
+    if (merge_success) {
       *status = Status::OK();
     } else {
       RecordTick(db_statistics_, NUMBER_MERGE_FAILURES);