Optimize GetLatestSequenceForKey

Summary: DBImpl::GetLatestSequenceForKey() can do memcpy's to load a value that will never be used.  This can be optimized by changing all the Get() functions called to optionally not fetch the value (and only fetch the sequencenumber).

Test Plan: optimistic_transaction_test and transaction_test

Reviewers: anthony

Reviewed By: anthony

Subscribers: leveldb, dhruba, hermanlee4

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

db/db_impl.cc

@@ -5686,7 +5686,6 @@ Status DBImpl::GetLatestSequenceForKey(SuperVersion* sv, const Slice& key,
                                        bool cache_only, SequenceNumber* seq,
                                        bool* found_record_for_key) {
   Status s;
-  std::string value;
   MergeContext merge_context;
 
   SequenceNumber current_seq = versions_->LastSequence();
@@ -5695,11 +5694,8 @@ Status DBImpl::GetLatestSequenceForKey(SuperVersion* sv, const Slice& key,
   *seq = kMaxSequenceNumber;
   *found_record_for_key = false;
 
-  // TODO(agiardullo): Should optimize all the Get() functions below to not
-  // return a value since we do not use it.
-
   // Check if there is a record for this key in the latest memtable
-  sv->mem->Get(lkey, &value, &s, &merge_context, seq);
+  sv->mem->Get(lkey, nullptr, &s, &merge_context, seq);
 
   if (!(s.ok() || s.IsNotFound() || s.IsMergeInProgress())) {
     // unexpected error reading memtable.
@@ -5717,7 +5713,7 @@ Status DBImpl::GetLatestSequenceForKey(SuperVersion* sv, const Slice& key,
   }
 
   // Check if there is a record for this key in the immutable memtables
-  sv->imm->Get(lkey, &value, &s, &merge_context, seq);
+  sv->imm->Get(lkey, nullptr, &s, &merge_context, seq);
 
   if (!(s.ok() || s.IsNotFound() || s.IsMergeInProgress())) {
     // unexpected error reading memtable.
@@ -5735,7 +5731,7 @@ Status DBImpl::GetLatestSequenceForKey(SuperVersion* sv, const Slice& key,
   }
 
   // Check if there is a record for this key in the immutable memtables
-  sv->imm->GetFromHistory(lkey, &value, &s, &merge_context, seq);
+  sv->imm->GetFromHistory(lkey, nullptr, &s, &merge_context, seq);
 
   if (!(s.ok() || s.IsNotFound() || s.IsMergeInProgress())) {
     // unexpected error reading memtable.
@@ -5758,7 +5754,7 @@ Status DBImpl::GetLatestSequenceForKey(SuperVersion* sv, const Slice& key,
     // Check tables
     ReadOptions read_options;
 
-    sv->current->Get(read_options, lkey, &value, &s, &merge_context,
+    sv->current->Get(read_options, lkey, nullptr, &s, &merge_context,
                      nullptr /* value_found */, found_record_for_key, seq);
 
     if (!(s.ok() || s.IsNotFound() || s.IsMergeInProgress())) {

db/memtable.cc

@@ -500,7 +500,7 @@ static bool SaveValue(void* arg, const char* entry) {
             *(s->status) =
                 Status::Corruption("Error: Could not perform merge.");
           }
-        } else {
+        } else if (s->value != nullptr) {
           s->value->assign(v.data(), v.size());
         }
         if (s->inplace_update_support) {

table/get_context.cc

@@ -70,8 +70,10 @@ void GetContext::SaveValue(const Slice& value, SequenceNumber seq) {
   appendToReplayLog(replay_log_, kTypeValue, value);
 
   state_ = kFound;
+  if (value_ != nullptr) {
     value_->assign(value.data(), value.size());
   }
+}
 
 bool GetContext::SaveValue(const ParsedInternalKey& parsed_key,
                            const Slice& value) {
@@ -93,10 +95,13 @@ bool GetContext::SaveValue(const ParsedInternalKey& parsed_key,
         assert(state_ == kNotFound || state_ == kMerge);
         if (kNotFound == state_) {
           state_ = kFound;
+          if (value_ != nullptr) {
             value_->assign(value.data(), value.size());
+          }
         } else if (kMerge == state_) {
           assert(merge_operator_ != nullptr);
           state_ = kFound;
+          if (value_ != nullptr) {
             bool merge_success = false;
             {
               StopWatchNano timer(env_, statistics_ != nullptr);
@@ -112,6 +117,7 @@ bool GetContext::SaveValue(const ParsedInternalKey& parsed_key,
               state_ = kCorrupt;
             }
           }
+        }
         return false;
 
       case kTypeDeletion:
@@ -123,6 +129,7 @@ bool GetContext::SaveValue(const ParsedInternalKey& parsed_key,
           state_ = kDeleted;
         } else if (kMerge == state_) {
           state_ = kFound;
+          if (value_ != nullptr) {
             bool merge_success = false;
             {
               StopWatchNano timer(env_, statistics_ != nullptr);
@@ -138,6 +145,7 @@ bool GetContext::SaveValue(const ParsedInternalKey& parsed_key,
               state_ = kCorrupt;
             }
           }
+        }
         return false;
 
       case kTypeMerge: