update IterKey that can get user key and internal key explicitly

Summary:
to void future bug that caused by the mix of userkey/internalkey
Closes https://github.com/facebook/rocksdb/pull/2084

Differential Revision: D4825889

Pulled By: lightmark

fbshipit-source-id: 28411db

15 files changed, 176 insertions, 139 deletions

diff --git a/db/compaction_iterator.cc b/db/compaction_iterator.cc
index a50572276..acbf1003c 100644
--- a/db/compaction_iterator.cc
+++ b/db/compaction_iterator.cc
@@ -105,7 +105,7 @@ void CompactionIterator::Next() {
       assert(valid_key);
       // Keep current_key_ in sync.
       current_key_.UpdateInternalKey(ikey_.sequence, ikey_.type);
-      key_ = current_key_.GetKey();
+      key_ = current_key_.GetInternalKey();
       ikey_.user_key = current_key_.GetUserKey();
       valid_ = true;
     } else {
@@ -151,7 +151,7 @@ void CompactionIterator::NextFromInput() {
         status_ = Status::Corruption("Corrupted internal key not expected.");
         break;
       }
-      key_ = current_key_.SetKey(key_);
+      key_ = current_key_.SetInternalKey(key_);
       has_current_user_key_ = false;
       current_user_key_sequence_ = kMaxSequenceNumber;
       current_user_key_snapshot_ = 0;
@@ -181,7 +181,7 @@ void CompactionIterator::NextFromInput() {
         !cmp_->Equal(ikey_.user_key, current_user_key_)) {
       // First occurrence of this user key
       // Copy key for output
-      key_ = current_key_.SetKey(key_, &ikey_);
+      key_ = current_key_.SetInternalKey(key_, &ikey_);
       current_user_key_ = ikey_.user_key;
       has_current_user_key_ = true;
       has_outputted_key_ = false;
@@ -241,7 +241,7 @@ void CompactionIterator::NextFromInput() {
       // Need to have the compaction filter process multiple versions
       // if we have versions on both sides of a snapshot
       current_key_.UpdateInternalKey(ikey_.sequence, ikey_.type);
-      key_ = current_key_.GetKey();
+      key_ = current_key_.GetInternalKey();
       ikey_.user_key = current_key_.GetUserKey();
     }
 
@@ -453,7 +453,7 @@ void CompactionIterator::NextFromInput() {
         assert(valid_key);
         // Keep current_key_ in sync.
         current_key_.UpdateInternalKey(ikey_.sequence, ikey_.type);
-        key_ = current_key_.GetKey();
+        key_ = current_key_.GetInternalKey();
         ikey_.user_key = current_key_.GetUserKey();
         valid_ = true;
       } else {
diff --git a/db/compaction_job.cc b/db/compaction_job.cc
index dcfc7aa09..ecbce1e6e 100644
--- a/db/compaction_job.cc
+++ b/db/compaction_job.cc
@@ -733,7 +733,7 @@ void CompactionJob::ProcessKeyValueCompaction(SubcompactionState* sub_compact) {
   if (start != nullptr) {
     IterKey start_iter;
     start_iter.SetInternalKey(*start, kMaxSequenceNumber, kValueTypeForSeek);
-    input->Seek(start_iter.GetKey());
+    input->Seek(start_iter.GetInternalKey());
   } else {
     input->SeekToFirst();
   }
diff --git a/db/db_iter.cc b/db/db_iter.cc
index e8ec9edfa..20850cf9b 100644
--- a/db/db_iter.cc
+++ b/db/db_iter.cc
@@ -162,7 +162,7 @@ class DBIter: public Iterator {
   virtual bool Valid() const override { return valid_; }
   virtual Slice key() const override {
     assert(valid_);
-    return saved_key_.GetKey();
+    return saved_key_.GetUserKey();
   }
   virtual Slice value() const override {
     assert(valid_);
@@ -406,7 +406,8 @@ void DBIter::FindNextUserEntryInternal(bool skipping, bool prefix_check) {
 
     if (ikey.sequence <= sequence_) {
       if (skipping &&
-          user_comparator_->Compare(ikey.user_key, saved_key_.GetKey()) <= 0) {
+          user_comparator_->Compare(ikey.user_key, saved_key_.GetUserKey()) <=
+              0) {
         num_skipped++;  // skip this entry
         PERF_COUNTER_ADD(internal_key_skipped_count, 1);
       } else {
@@ -416,14 +417,14 @@ void DBIter::FindNextUserEntryInternal(bool skipping, bool prefix_check) {
           case kTypeSingleDeletion:
             // Arrange to skip all upcoming entries for this key since
             // they are hidden by this deletion.
-            saved_key_.SetKey(
+            saved_key_.SetUserKey(
                 ikey.user_key,
                 !iter_->IsKeyPinned() || !pin_thru_lifetime_ /* copy */);
             skipping = true;
             PERF_COUNTER_ADD(internal_delete_skipped_count, 1);
             break;
           case kTypeValue:
-            saved_key_.SetKey(
+            saved_key_.SetUserKey(
                 ikey.user_key,
                 !iter_->IsKeyPinned() || !pin_thru_lifetime_ /* copy */);
             if (range_del_agg_.ShouldDelete(
@@ -440,7 +441,7 @@ void DBIter::FindNextUserEntryInternal(bool skipping, bool prefix_check) {
             }
             break;
           case kTypeMerge:
-            saved_key_.SetKey(
+            saved_key_.SetUserKey(
                 ikey.user_key,
                 !iter_->IsKeyPinned() || !pin_thru_lifetime_ /* copy */);
             if (range_del_agg_.ShouldDelete(
@@ -471,12 +472,13 @@ void DBIter::FindNextUserEntryInternal(bool skipping, bool prefix_check) {
 
       // Here saved_key_ may contain some old key, or the default empty key, or
       // key assigned by some random other method. We don't care.
-      if (user_comparator_->Compare(ikey.user_key, saved_key_.GetKey()) <= 0) {
+      if (user_comparator_->Compare(ikey.user_key, saved_key_.GetUserKey()) <=
+          0) {
         num_skipped++;
       } else {
-        saved_key_.SetKey(
-                ikey.user_key,
-                !iter_->IsKeyPinned() || !pin_thru_lifetime_ /* copy */);
+        saved_key_.SetUserKey(
+            ikey.user_key,
+            !iter_->IsKeyPinned() || !pin_thru_lifetime_ /* copy */);
         skipping = false;
         num_skipped = 0;
       }
@@ -491,8 +493,8 @@ void DBIter::FindNextUserEntryInternal(bool skipping, bool prefix_check) {
         // We're looking for the next user-key but all we see are the same
         // user-key with decreasing sequence numbers. Fast forward to
         // sequence number 0 and type deletion (the smallest type).
-        AppendInternalKey(&last_key, ParsedInternalKey(saved_key_.GetKey(), 0,
-                                                       kTypeDeletion));
+        AppendInternalKey(&last_key, ParsedInternalKey(saved_key_.GetUserKey(),
+                                                       0, kTypeDeletion));
         // Don't set skipping = false because we may still see more user-keys
         // equal to saved_key_.
       } else {
@@ -502,7 +504,7 @@ void DBIter::FindNextUserEntryInternal(bool skipping, bool prefix_check) {
         // many times since our snapshot was taken, not the case when a lot of
         // different keys were inserted after our snapshot was taken.
         AppendInternalKey(&last_key,
-                          ParsedInternalKey(saved_key_.GetKey(), sequence_,
+                          ParsedInternalKey(saved_key_.GetUserKey(), sequence_,
                                             kValueTypeForSeek));
       }
       iter_->Seek(last_key);
@@ -543,7 +545,7 @@ void DBIter::MergeValuesNewToOld() {
       continue;
     }
 
-    if (!user_comparator_->Equal(ikey.user_key, saved_key_.GetKey())) {
+    if (!user_comparator_->Equal(ikey.user_key, saved_key_.GetUserKey())) {
       // hit the next user key, stop right here
       break;
     } else if (kTypeDeletion == ikey.type || kTypeSingleDeletion == ikey.type ||
@@ -583,9 +585,10 @@ void DBIter::MergeValuesNewToOld() {
   // a deletion marker.
   // feed null as the existing value to the merge operator, such that
   // client can differentiate this scenario and do things accordingly.
-  s = MergeHelper::TimedFullMerge(merge_operator_, saved_key_.GetKey(), nullptr,
-                                  merge_context_.GetOperands(), &saved_value_,
-                                  logger_, statistics_, env_, &pinned_value_);
+  s = MergeHelper::TimedFullMerge(merge_operator_, saved_key_.GetUserKey(),
+                                  nullptr, merge_context_.GetOperands(),
+                                  &saved_value_, logger_, statistics_, env_,
+                                  &pinned_value_);
   if (!s.ok()) {
     status_ = s;
   }
@@ -612,8 +615,8 @@ void DBIter::ReverseToForward() {
   if (prefix_extractor_ != nullptr && !total_order_seek_) {
     IterKey last_key;
     last_key.SetInternalKey(ParsedInternalKey(
-        saved_key_.GetKey(), kMaxSequenceNumber, kValueTypeForSeek));
-    iter_->Seek(last_key.GetKey());
+        saved_key_.GetUserKey(), kMaxSequenceNumber, kValueTypeForSeek));
+    iter_->Seek(last_key.GetInternalKey());
   }
   FindNextUserKey();
   direction_ = kForward;
@@ -626,9 +629,9 @@ void DBIter::ReverseToForward() {
 void DBIter::ReverseToBackward() {
   if (prefix_extractor_ != nullptr && !total_order_seek_) {
     IterKey last_key;
-    last_key.SetInternalKey(
-        ParsedInternalKey(saved_key_.GetKey(), 0, kValueTypeForSeekForPrev));
-    iter_->SeekForPrev(last_key.GetKey());
+    last_key.SetInternalKey(ParsedInternalKey(saved_key_.GetUserKey(), 0,
+                                              kValueTypeForSeekForPrev));
+    iter_->SeekForPrev(last_key.GetInternalKey());
   }
   if (current_entry_is_merged_) {
     // Not placed in the same key. Need to call Prev() until finding the
@@ -640,7 +643,8 @@ void DBIter::ReverseToBackward() {
     ParsedInternalKey ikey;
     FindParseableKey(&ikey, kReverse);
     while (iter_->Valid() &&
-           user_comparator_->Compare(ikey.user_key, saved_key_.GetKey()) > 0) {
+           user_comparator_->Compare(ikey.user_key, saved_key_.GetUserKey()) >
+               0) {
       if (ikey.sequence > sequence_) {
         PERF_COUNTER_ADD(internal_recent_skipped_count, 1);
       } else {
@@ -654,7 +658,8 @@ void DBIter::ReverseToBackward() {
   if (iter_->Valid()) {
     ParsedInternalKey ikey;
     assert(ParseKey(&ikey));
-    assert(user_comparator_->Compare(ikey.user_key, saved_key_.GetKey()) <= 0);
+    assert(user_comparator_->Compare(ikey.user_key, saved_key_.GetUserKey()) <=
+           0);
   }
 #endif
 
@@ -671,8 +676,9 @@ void DBIter::PrevInternal() {
   ParsedInternalKey ikey;
 
   while (iter_->Valid()) {
-    saved_key_.SetKey(ExtractUserKey(iter_->key()),
-                      !iter_->IsKeyPinned() || !pin_thru_lifetime_ /* copy */);
+    saved_key_.SetUserKey(
+        ExtractUserKey(iter_->key()),
+        !iter_->IsKeyPinned() || !pin_thru_lifetime_ /* copy */);
 
     if (FindValueForCurrentKey()) {
       valid_ = true;
@@ -680,11 +686,11 @@ void DBIter::PrevInternal() {
         return;
       }
       FindParseableKey(&ikey, kReverse);
-      if (user_comparator_->Equal(ikey.user_key, saved_key_.GetKey())) {
+      if (user_comparator_->Equal(ikey.user_key, saved_key_.GetUserKey())) {
         FindPrevUserKey();
       }
       if (valid_ && prefix_extractor_ && prefix_same_as_start_ &&
-          prefix_extractor_->Transform(saved_key_.GetKey())
+          prefix_extractor_->Transform(saved_key_.GetUserKey())
                   .compare(prefix_start_key_) != 0) {
         valid_ = false;
       }
@@ -699,7 +705,7 @@ void DBIter::PrevInternal() {
       break;
     }
     FindParseableKey(&ikey, kReverse);
-    if (user_comparator_->Equal(ikey.user_key, saved_key_.GetKey())) {
+    if (user_comparator_->Equal(ikey.user_key, saved_key_.GetUserKey())) {
       FindPrevUserKey();
     }
   }
@@ -729,7 +735,7 @@ bool DBIter::FindValueForCurrentKey() {
   TempPinData();
   size_t num_skipped = 0;
   while (iter_->Valid() && ikey.sequence <= sequence_ &&
-         user_comparator_->Equal(ikey.user_key, saved_key_.GetKey())) {
+         user_comparator_->Equal(ikey.user_key, saved_key_.GetUserKey())) {
     if (TooManyInternalKeysSkipped()) {
       return false;
     }
@@ -780,7 +786,7 @@ bool DBIter::FindValueForCurrentKey() {
     }
 
     PERF_COUNTER_ADD(internal_key_skipped_count, 1);
-    assert(user_comparator_->Equal(ikey.user_key, saved_key_.GetKey()));
+    assert(user_comparator_->Equal(ikey.user_key, saved_key_.GetUserKey()));
     iter_->Prev();
     ++num_skipped;
     FindParseableKey(&ikey, kReverse);
@@ -798,14 +804,14 @@ bool DBIter::FindValueForCurrentKey() {
       if (last_not_merge_type == kTypeDeletion ||
           last_not_merge_type == kTypeSingleDeletion ||
           last_not_merge_type == kTypeRangeDeletion) {
-        s = MergeHelper::TimedFullMerge(merge_operator_, saved_key_.GetKey(),
-                                        nullptr, merge_context_.GetOperands(),
-                                        &saved_value_, logger_, statistics_,
-                                        env_, &pinned_value_);
+        s = MergeHelper::TimedFullMerge(
+            merge_operator_, saved_key_.GetUserKey(), nullptr,
+            merge_context_.GetOperands(), &saved_value_, logger_, statistics_,
+            env_, &pinned_value_);
       } else {
         assert(last_not_merge_type == kTypeValue);
         s = MergeHelper::TimedFullMerge(
-            merge_operator_, saved_key_.GetKey(), &pinned_value_,
+            merge_operator_, saved_key_.GetUserKey(), &pinned_value_,
             merge_context_.GetOperands(), &saved_value_, logger_, statistics_,
             env_, &pinned_value_);
       }
@@ -831,8 +837,8 @@ bool DBIter::FindValueForCurrentKeyUsingSeek() {
   // FindValueForCurrentKeyUsingSeek()
   assert(pinned_iters_mgr_.PinningEnabled());
   std::string last_key;
-  AppendInternalKey(&last_key, ParsedInternalKey(saved_key_.GetKey(), sequence_,
-                                                 kValueTypeForSeek));
+  AppendInternalKey(&last_key, ParsedInternalKey(saved_key_.GetUserKey(),
+                                                 sequence_, kValueTypeForSeek));
   iter_->Seek(last_key);
   RecordTick(statistics_, NUMBER_OF_RESEEKS_IN_ITERATION);
 
@@ -859,7 +865,7 @@ bool DBIter::FindValueForCurrentKeyUsingSeek() {
   merge_context_.Clear();
   while (
       iter_->Valid() &&
-      user_comparator_->Equal(ikey.user_key, saved_key_.GetKey()) &&
+      user_comparator_->Equal(ikey.user_key, saved_key_.GetUserKey()) &&
       ikey.type == kTypeMerge &&
       !range_del_agg_.ShouldDelete(
           ikey, RangeDelAggregator::RangePositioningMode::kBackwardTraversal)) {
@@ -872,17 +878,17 @@ bool DBIter::FindValueForCurrentKeyUsingSeek() {
 
   Status s;
   if (!iter_->Valid() ||
-      !user_comparator_->Equal(ikey.user_key, saved_key_.GetKey()) ||
+      !user_comparator_->Equal(ikey.user_key, saved_key_.GetUserKey()) ||
       ikey.type == kTypeDeletion || ikey.type == kTypeSingleDeletion ||
       range_del_agg_.ShouldDelete(
           ikey, RangeDelAggregator::RangePositioningMode::kBackwardTraversal)) {
-    s = MergeHelper::TimedFullMerge(merge_operator_, saved_key_.GetKey(),
+    s = MergeHelper::TimedFullMerge(merge_operator_, saved_key_.GetUserKey(),
                                     nullptr, merge_context_.GetOperands(),
                                     &saved_value_, logger_, statistics_, env_,
                                     &pinned_value_);
     // Make iter_ valid and point to saved_key_
     if (!iter_->Valid() ||
-        !user_comparator_->Equal(ikey.user_key, saved_key_.GetKey())) {
+        !user_comparator_->Equal(ikey.user_key, saved_key_.GetUserKey())) {
       iter_->Seek(last_key);
       RecordTick(statistics_, NUMBER_OF_RESEEKS_IN_ITERATION);
     }
@@ -894,9 +900,10 @@ bool DBIter::FindValueForCurrentKeyUsingSeek() {
   }
 
   const Slice& val = iter_->value();
-  s = MergeHelper::TimedFullMerge(merge_operator_, saved_key_.GetKey(), &val,
-                                  merge_context_.GetOperands(), &saved_value_,
-                                  logger_, statistics_, env_, &pinned_value_);
+  s = MergeHelper::TimedFullMerge(merge_operator_, saved_key_.GetUserKey(),
+                                  &val, merge_context_.GetOperands(),
+                                  &saved_value_, logger_, statistics_, env_,
+                                  &pinned_value_);
   valid_ = true;
   if (!s.ok()) {
     status_ = s;
@@ -915,7 +922,7 @@ void DBIter::FindNextUserKey() {
   ParsedInternalKey ikey;
   FindParseableKey(&ikey, kForward);
   while (iter_->Valid() &&
-         !user_comparator_->Equal(ikey.user_key, saved_key_.GetKey())) {
+         !user_comparator_->Equal(ikey.user_key, saved_key_.GetUserKey())) {
     iter_->Next();
     FindParseableKey(&ikey, kForward);
   }
@@ -930,9 +937,10 @@ void DBIter::FindPrevUserKey() {
   ParsedInternalKey ikey;
   FindParseableKey(&ikey, kReverse);
   int cmp;
-  while (iter_->Valid() && ((cmp = user_comparator_->Compare(
-                                 ikey.user_key, saved_key_.GetKey())) == 0 ||
-                            (cmp > 0 && ikey.sequence > sequence_))) {
+  while (iter_->Valid() &&
+         ((cmp = user_comparator_->Compare(ikey.user_key,
+                                           saved_key_.GetUserKey())) == 0 ||
+          (cmp > 0 && ikey.sequence > sequence_))) {
     if (TooManyInternalKeysSkipped()) {
       return;
     }
@@ -942,8 +950,8 @@ void DBIter::FindPrevUserKey() {
         num_skipped = 0;
         IterKey last_key;
         last_key.SetInternalKey(ParsedInternalKey(
-            saved_key_.GetKey(), kMaxSequenceNumber, kValueTypeForSeek));
-        iter_->Seek(last_key.GetKey());
+            saved_key_.GetUserKey(), kMaxSequenceNumber, kValueTypeForSeek));
+        iter_->Seek(last_key.GetInternalKey());
         RecordTick(statistics_, NUMBER_OF_RESEEKS_IN_ITERATION);
       } else {
         ++num_skipped;
@@ -991,7 +999,7 @@ void DBIter::Seek(const Slice& target) {
 
   {
     PERF_TIMER_GUARD(seek_internal_seek_time);
-    iter_->Seek(saved_key_.GetKey());
+    iter_->Seek(saved_key_.GetInternalKey());
     range_del_agg_.InvalidateTombstoneMapPositions();
   }
   RecordTick(statistics_, NUMBER_DB_SEEK);
@@ -1001,9 +1009,6 @@ void DBIter::Seek(const Slice& target) {
     }
     direction_ = kForward;
     ClearSavedValue();
-    // convert the InternalKey to UserKey in saved_key_ before
-    // passed to FindNextUserEntry
-    saved_key_.Reserve(saved_key_.Size() - 8);
     FindNextUserEntry(false /* not skipping */, prefix_same_as_start_);
     if (!valid_) {
       prefix_start_key_.clear();
@@ -1019,8 +1024,8 @@ void DBIter::Seek(const Slice& target) {
   }
 
   if (valid_ && prefix_extractor_ && prefix_same_as_start_) {
-    prefix_start_buf_.SetKey(prefix_start_key_);
-    prefix_start_key_ = prefix_start_buf_.GetKey();
+    prefix_start_buf_.SetUserKey(prefix_start_key_);
+    prefix_start_key_ = prefix_start_buf_.GetUserKey();
   }
 }
 
@@ -1035,7 +1040,7 @@ void DBIter::SeekForPrev(const Slice& target) {
 
   {
     PERF_TIMER_GUARD(seek_internal_seek_time);
-    iter_->SeekForPrev(saved_key_.GetKey());
+    iter_->SeekForPrev(saved_key_.GetInternalKey());
     range_del_agg_.InvalidateTombstoneMapPositions();
   }
 
@@ -1060,8 +1065,8 @@ void DBIter::SeekForPrev(const Slice& target) {
     valid_ = false;
   }
   if (valid_ && prefix_extractor_ && prefix_same_as_start_) {
-    prefix_start_buf_.SetKey(prefix_start_key_);
-    prefix_start_key_ = prefix_start_buf_.GetKey();
+    prefix_start_buf_.SetUserKey(prefix_start_key_);
+    prefix_start_key_ = prefix_start_buf_.GetUserKey();
   }
 }
 
@@ -1084,8 +1089,9 @@ void DBIter::SeekToFirst() {
 
   RecordTick(statistics_, NUMBER_DB_SEEK);
   if (iter_->Valid()) {
-    saved_key_.SetKey(ExtractUserKey(iter_->key()),
-                      !iter_->IsKeyPinned() || !pin_thru_lifetime_ /* copy */);
+    saved_key_.SetUserKey(
+        ExtractUserKey(iter_->key()),
+        !iter_->IsKeyPinned() || !pin_thru_lifetime_ /* copy */);
     FindNextUserEntry(false /* not skipping */, false /* no prefix check */);
     if (statistics_ != nullptr) {
       if (valid_) {
@@ -1097,8 +1103,9 @@ void DBIter::SeekToFirst() {
     valid_ = false;
   }
   if (valid_ && prefix_extractor_ && prefix_same_as_start_) {
-    prefix_start_buf_.SetKey(prefix_extractor_->Transform(saved_key_.GetKey()));
-    prefix_start_key_ = prefix_start_buf_.GetKey();
+    prefix_start_buf_.SetUserKey(
+        prefix_extractor_->Transform(saved_key_.GetUserKey()));
+    prefix_start_key_ = prefix_start_buf_.GetUserKey();
   }
 }
 
@@ -1140,8 +1147,9 @@ void DBIter::SeekToLast() {
     }
   }
   if (valid_ && prefix_extractor_ && prefix_same_as_start_) {
-    prefix_start_buf_.SetKey(prefix_extractor_->Transform(saved_key_.GetKey()));
-    prefix_start_key_ = prefix_start_buf_.GetKey();
+    prefix_start_buf_.SetUserKey(
+        prefix_extractor_->Transform(saved_key_.GetUserKey()));
+    prefix_start_key_ = prefix_start_buf_.GetUserKey();
   }
 }
 
diff --git a/db/dbformat.h b/db/dbformat.h
index e4fc29f60..b4dbeb2ab 100644
--- a/db/dbformat.h
+++ b/db/dbformat.h
@@ -303,15 +303,26 @@ inline LookupKey::~LookupKey() {
 class IterKey {
  public:
   IterKey()
-      : buf_(space_), buf_size_(sizeof(space_)), key_(buf_), key_size_(0) {}
+      : buf_(space_),
+        buf_size_(sizeof(space_)),
+        key_(buf_),
+        key_size_(0),
+        is_user_key_(true) {}
 
   ~IterKey() { ResetBuffer(); }
 
-  Slice GetKey() const { return Slice(key_, key_size_); }
+  Slice GetInternalKey() const {
+    assert(!IsUserKey());
+    return Slice(key_, key_size_);
+  }
 
   Slice GetUserKey() const {
-    assert(key_size_ >= 8);
-    return Slice(key_, key_size_ - 8);
+    if (IsUserKey()) {
+      return Slice(key_, key_size_);
+    } else {
+      assert(key_size_ >= 8);
+      return Slice(key_, key_size_ - 8);
+    }
   }
 
   size_t Size() const { return key_size_; }
@@ -349,27 +360,22 @@ class IterKey {
     key_size_ = total_size;
   }
 
-  Slice SetKey(const Slice& key, bool copy = true) {
-    size_t size = key.size();
-    if (copy) {
-      // Copy key to buf_
-      EnlargeBufferIfNeeded(size);
-      memcpy(buf_, key.data(), size);
-      key_ = buf_;
-    } else {
-      // Update key_ to point to external memory
-      key_ = key.data();
-    }
-    key_size_ = size;
-    return Slice(key_, key_size_);
+  Slice SetUserKey(const Slice& key, bool copy = true) {
+    is_user_key_ = true;
+    return SetKeyImpl(key, copy);
+  }
+
+  Slice SetInternalKey(const Slice& key, bool copy = true) {
+    is_user_key_ = false;
+    return SetKeyImpl(key, copy);
   }
 
   // Copies the content of key, updates the reference to the user key in ikey
   // and returns a Slice referencing the new copy.
-  Slice SetKey(const Slice& key, ParsedInternalKey* ikey) {
+  Slice SetInternalKey(const Slice& key, ParsedInternalKey* ikey) {
     size_t key_n = key.size();
     assert(key_n >= 8);
-    SetKey(key);
+    SetInternalKey(key);
     ikey->user_key = Slice(key_, key_n - 8);
     return Slice(key_, key_n);
   }
@@ -408,6 +414,7 @@ class IterKey {
 
     key_ = buf_;
     key_size_ = psize + usize + sizeof(uint64_t);
+    is_user_key_ = false;
   }
 
   void SetInternalKey(const Slice& user_key, SequenceNumber s,
@@ -436,14 +443,33 @@ class IterKey {
     char* ptr = EncodeVarint32(buf_, static_cast<uint32_t>(size));
     memcpy(ptr, key.data(), size);
     key_ = buf_;
+    is_user_key_ = true;
   }
 
+  bool IsUserKey() const { return is_user_key_; }
+
  private:
   char* buf_;
   size_t buf_size_;
   const char* key_;
   size_t key_size_;
   char space_[32];  // Avoid allocation for short keys
+  bool is_user_key_;
+
+  Slice SetKeyImpl(const Slice& key, bool copy) {
+    size_t size = key.size();
+    if (copy) {
+      // Copy key to buf_
+      EnlargeBufferIfNeeded(size);
+      memcpy(buf_, key.data(), size);
+      key_ = buf_;
+    } else {
+      // Update key_ to point to external memory
+      key_ = key.data();
+    }
+    key_size_ = size;
+    return Slice(key_, key_size_);
+  }
 
   void ResetBuffer() {
     if (buf_ != space_) {
diff --git a/db/dbformat_test.cc b/db/dbformat_test.cc
index b431690cc..d35591cca 100644
--- a/db/dbformat_test.cc
+++ b/db/dbformat_test.cc
@@ -139,38 +139,38 @@ TEST_F(FormatTest, IterKeyOperation) {
   const char p[] = "abcdefghijklmnopqrstuvwxyz";
   const char q[] = "0123456789";
 
-  ASSERT_EQ(std::string(k.GetKey().data(), k.GetKey().size()),
+  ASSERT_EQ(std::string(k.GetUserKey().data(), k.GetUserKey().size()),
             std::string(""));
 
   k.TrimAppend(0, p, 3);
-  ASSERT_EQ(std::string(k.GetKey().data(), k.GetKey().size()),
+  ASSERT_EQ(std::string(k.GetUserKey().data(), k.GetUserKey().size()),
             std::string("abc"));
 
   k.TrimAppend(1, p, 3);
-  ASSERT_EQ(std::string(k.GetKey().data(), k.GetKey().size()),
+  ASSERT_EQ(std::string(k.GetUserKey().data(), k.GetUserKey().size()),
             std::string("aabc"));
 
   k.TrimAppend(0, p, 26);
-  ASSERT_EQ(std::string(k.GetKey().data(), k.GetKey().size()),
+  ASSERT_EQ(std::string(k.GetUserKey().data(), k.GetUserKey().size()),
             std::string("abcdefghijklmnopqrstuvwxyz"));
 
   k.TrimAppend(26, q, 10);
-  ASSERT_EQ(std::string(k.GetKey().data(), k.GetKey().size()),
+  ASSERT_EQ(std::string(k.GetUserKey().data(), k.GetUserKey().size()),
             std::string("abcdefghijklmnopqrstuvwxyz0123456789"));
 
   k.TrimAppend(36, q, 1);
-  ASSERT_EQ(std::string(k.GetKey().data(), k.GetKey().size()),
+  ASSERT_EQ(std::string(k.GetUserKey().data(), k.GetUserKey().size()),
             std::string("abcdefghijklmnopqrstuvwxyz01234567890"));
 
   k.TrimAppend(26, q, 1);
-  ASSERT_EQ(std::string(k.GetKey().data(), k.GetKey().size()),
+  ASSERT_EQ(std::string(k.GetUserKey().data(), k.GetUserKey().size()),
             std::string("abcdefghijklmnopqrstuvwxyz0"));
 
   // Size going up, memory allocation is triggered
   k.TrimAppend(27, p, 26);
-  ASSERT_EQ(std::string(k.GetKey().data(), k.GetKey().size()),
+  ASSERT_EQ(std::string(k.GetUserKey().data(), k.GetUserKey().size()),
             std::string("abcdefghijklmnopqrstuvwxyz0"
-              "abcdefghijklmnopqrstuvwxyz"));
+                        "abcdefghijklmnopqrstuvwxyz"));
 }
 
 TEST_F(FormatTest, UpdateInternalKey) {
diff --git a/db/forward_iterator.cc b/db/forward_iterator.cc
index 8eae808e8..7b5fd28d5 100644
--- a/db/forward_iterator.cc
+++ b/db/forward_iterator.cc
@@ -310,7 +310,7 @@ void ForwardIterator::SeekInternal(const Slice& internal_key,
             seek_to_first ||
             // prev_key_ > internal_key
             cfd_->internal_comparator().InternalKeyComparator::Compare(
-                prev_key_.GetKey(), internal_key) > 0)) {
+                prev_key_.GetInternalKey(), internal_key) > 0)) {
       // Some iterators are trimmed. Need to rebuild.
       RebuildIterators(true);
       // Already seeked mutable iter, so seek again
@@ -410,7 +410,7 @@ void ForwardIterator::SeekInternal(const Slice& internal_key,
     if (seek_to_first) {
       is_prev_set_ = false;
     } else {
-      prev_key_.SetKey(internal_key);
+      prev_key_.SetInternalKey(internal_key);
       is_prev_set_ = true;
       is_prev_inclusive_ = true;
     }
@@ -449,15 +449,15 @@ void ForwardIterator::Next() {
     if (is_prev_set_ && prefix_extractor_) {
       // advance prev_key_ to current_ only if they share the same prefix
       update_prev_key =
-        prefix_extractor_->Transform(prev_key_.GetKey()).compare(
-          prefix_extractor_->Transform(current_->key())) == 0;
+          prefix_extractor_->Transform(prev_key_.GetUserKey())
+              .compare(prefix_extractor_->Transform(current_->key())) == 0;
     } else {
       update_prev_key = true;
     }
 
 
     if (update_prev_key) {
-      prev_key_.SetKey(current_->key());
+      prev_key_.SetInternalKey(current_->key());
       is_prev_set_ = true;
       is_prev_inclusive_ = false;
     }
@@ -476,7 +476,7 @@ void ForwardIterator::Next() {
         current_ = nullptr;
       }
       if (update_prev_key) {
-        mutable_iter_->Seek(prev_key_.GetKey());
+        mutable_iter_->Seek(prev_key_.GetInternalKey());
       }
     }
   }
@@ -776,7 +776,7 @@ bool ForwardIterator::NeedToSeekImmutable(const Slice& target) {
   if (!valid_ || !current_ || !is_prev_set_ || !immutable_status_.ok()) {
     return true;
   }
-  Slice prev_key = prev_key_.GetKey();
+  Slice prev_key = prev_key_.GetInternalKey();
   if (prefix_extractor_ && prefix_extractor_->Transform(target).compare(
     prefix_extractor_->Transform(prev_key)) != 0) {
     return true;
diff --git a/db/managed_iterator.cc b/db/managed_iterator.cc
index c9ded718f..31ddf841c 100644
--- a/db/managed_iterator.cc
+++ b/db/managed_iterator.cc
@@ -196,12 +196,12 @@ void ManagedIterator::Next() {
 
 Slice ManagedIterator::key() const {
   assert(valid_);
-  return cached_key_.GetKey();
+  return cached_key_.GetUserKey();
 }
 
 Slice ManagedIterator::value() const {
   assert(valid_);
-  return cached_value_.GetKey();
+  return cached_value_.GetUserKey();
 }
 
 Status ManagedIterator::status() const { return status_; }
@@ -221,8 +221,8 @@ void ManagedIterator::UpdateCurrent() {
   }
 
   status_ = Status::OK();
-  cached_key_.SetKey(mutable_iter_->key());
-  cached_value_.SetKey(mutable_iter_->value());
+  cached_key_.SetUserKey(mutable_iter_->key());
+  cached_value_.SetUserKey(mutable_iter_->value());
 }
 
 void ManagedIterator::ReleaseIter(bool only_old) {
diff --git a/db/table_cache.cc b/db/table_cache.cc
index 7156da386..36e821517 100644
--- a/db/table_cache.cc
+++ b/db/table_cache.cc
@@ -286,7 +286,7 @@ Status TableCache::Get(const ReadOptions& options,
                              user_key.size());
 
     if (auto row_handle =
-            ioptions_.row_cache->Lookup(row_cache_key.GetKey())) {
+            ioptions_.row_cache->Lookup(row_cache_key.GetUserKey())) {
       auto found_row_cache_entry = static_cast<const std::string*>(
           ioptions_.row_cache->Value(row_handle));
       replayGetContextLog(*found_row_cache_entry, user_key, get_context);
@@ -343,7 +343,7 @@ Status TableCache::Get(const ReadOptions& options,
     size_t charge =
         row_cache_key.Size() + row_cache_entry->size() + sizeof(std::string);
     void* row_ptr = new std::string(std::move(*row_cache_entry));
-    ioptions_.row_cache->Insert(row_cache_key.GetKey(), row_ptr, charge,
+    ioptions_.row_cache->Insert(row_cache_key.GetUserKey(), row_ptr, charge,
                                 &DeleteEntry<std::string>);
   }
 #endif  // ROCKSDB_LITE
diff --git a/memtable/hash_linklist_rep.cc b/memtable/hash_linklist_rep.cc
index 7caf66531..df2cf6cef 100644
--- a/memtable/hash_linklist_rep.cc
+++ b/memtable/hash_linklist_rep.cc
@@ -433,7 +433,7 @@ class HashLinkListRep : public MemTableRep {
         } else {
           IterKey encoded_key;
           encoded_key.EncodeLengthPrefixedKey(k);
-          skip_list_iter_->Seek(encoded_key.GetKey().data());
+          skip_list_iter_->Seek(encoded_key.GetUserKey().data());
         }
       } else {
         // The bucket is organized as a linked list
diff --git a/table/block.cc b/table/block.cc
index 6ef46f4ce..4bc0add86 100644
--- a/table/block.cc
+++ b/table/block.cc
@@ -87,7 +87,7 @@ void BlockIter::Prev() {
     const Slice current_key(key_ptr, current_prev_entry.key_size);
 
     current_ = current_prev_entry.offset;
-    key_.SetKey(current_key, false /* copy */);
+    key_.SetInternalKey(current_key, false /* copy */);
     value_ = current_prev_entry.value;
 
     return;
@@ -155,7 +155,7 @@ void BlockIter::Seek(const Slice& target) {
   // Linear search (within restart block) for first key >= target
 
   while (true) {
-    if (!ParseNextKey() || Compare(key_.GetKey(), target) >= 0) {
+    if (!ParseNextKey() || Compare(key_.GetInternalKey(), target) >= 0) {
       return;
     }
   }
@@ -176,12 +176,12 @@ void BlockIter::SeekForPrev(const Slice& target) {
   SeekToRestartPoint(index);
   // Linear search (within restart block) for first key >= target
 
-  while (ParseNextKey() && Compare(key_.GetKey(), target) < 0) {
+  while (ParseNextKey() && Compare(key_.GetInternalKey(), target) < 0) {
   }
   if (!Valid()) {
     SeekToLast();
   } else {
-    while (Valid() && Compare(key_.GetKey(), target) > 0) {
+    while (Valid() && Compare(key_.GetInternalKey(), target) > 0) {
       Prev();
     }
   }
@@ -234,7 +234,7 @@ bool BlockIter::ParseNextKey() {
     if (shared == 0) {
       // If this key dont share any bytes with prev key then we dont need
       // to decode it and can use it's address in the block directly.
-      key_.SetKey(Slice(p, non_shared), false /* copy */);
+      key_.SetInternalKey(Slice(p, non_shared), false /* copy */);
       key_pinned_ = true;
     } else {
       // This key share `shared` bytes with prev key, we need to decode it
@@ -246,8 +246,8 @@ bool BlockIter::ParseNextKey() {
       // If we are reading a file with a global sequence number we should
       // expect that all encoded sequence numbers are zeros and all value
       // types are kTypeValue
-      assert(GetInternalKeySeqno(key_.GetKey()) == 0);
-      assert(ExtractValueType(key_.GetKey()) == ValueType::kTypeValue);
+      assert(GetInternalKeySeqno(key_.GetInternalKey()) == 0);
+      assert(ExtractValueType(key_.GetInternalKey()) == ValueType::kTypeValue);
 
       if (key_pinned_) {
         // TODO(tec): Investigate updating the seqno in the loaded block
diff --git a/table/block.h b/table/block.h
index 02daa83b5..acd1fc79e 100644
--- a/table/block.h
+++ b/table/block.h
@@ -264,7 +264,7 @@ class BlockIter : public InternalIterator {
   virtual Status status() const override { return status_; }
   virtual Slice key() const override {
     assert(Valid());
-    return key_.GetKey();
+    return key_.GetInternalKey();
   }
   virtual Slice value() const override {
     assert(Valid());
diff --git a/table/cuckoo_table_builder_test.cc b/table/cuckoo_table_builder_test.cc
index ac5bed157..bc8514ff2 100644
--- a/table/cuckoo_table_builder_test.cc
+++ b/table/cuckoo_table_builder_test.cc
@@ -124,7 +124,7 @@ class CuckooBuilderTest : public testing::Test {
   std::string GetInternalKey(Slice user_key, bool zero_seqno) {
     IterKey ikey;
     ikey.SetInternalKey(user_key, zero_seqno ? 0 : 1000, kTypeValue);
-    return ikey.GetKey().ToString();
+    return ikey.GetInternalKey().ToString();
   }
 
   uint64_t NextPowOf2(uint64_t num) {
diff --git a/table/cuckoo_table_reader.cc b/table/cuckoo_table_reader.cc
index ba699b251..c2f00c25c 100644
--- a/table/cuckoo_table_reader.cc
+++ b/table/cuckoo_table_reader.cc
@@ -322,7 +322,7 @@ void CuckooTableIterator::PrepareKVAtCurrIdx() {
     curr_key_.SetInternalKey(Slice(offset, reader_->user_key_length_),
                              0, kTypeValue);
   } else {
-    curr_key_.SetKey(Slice(offset, reader_->key_length_));
+    curr_key_.SetInternalKey(Slice(offset, reader_->key_length_));
   }
   curr_value_ = Slice(offset + reader_->key_length_, reader_->value_length_);
 }
@@ -352,7 +352,7 @@ void CuckooTableIterator::Prev() {
 
 Slice CuckooTableIterator::key() const {
   assert(Valid());
-  return curr_key_.GetKey();
+  return curr_key_.GetInternalKey();
 }
 
 Slice CuckooTableIterator::value() const {
diff --git a/table/cuckoo_table_reader_test.cc b/table/cuckoo_table_reader_test.cc
index 5d82d58fc..953aadc72 100644
--- a/table/cuckoo_table_reader_test.cc
+++ b/table/cuckoo_table_reader_test.cc
@@ -376,7 +376,7 @@ void GetKeys(uint64_t num, std::vector<std::string>* keys) {
   keys->clear();
   IterKey k;
   k.SetInternalKey("", 0, kTypeValue);
-  std::string internal_key_suffix = k.GetKey().ToString();
+  std::string internal_key_suffix = k.GetInternalKey().ToString();
   ASSERT_EQ(static_cast<size_t>(8), internal_key_suffix.size());
   for (uint64_t key_idx = 0; key_idx < num; ++key_idx) {
     uint64_t value = 2 * key_idx;
diff --git a/table/plain_table_key_coding.cc b/table/plain_table_key_coding.cc
index 8442f1129..a64a17238 100644
--- a/table/plain_table_key_coding.cc
+++ b/table/plain_table_key_coding.cc
@@ -112,10 +112,10 @@ Status PlainTableKeyEncoder::AppendKey(const Slice& key,
 
     Slice prefix =
         prefix_extractor_->Transform(Slice(key.data(), user_key_size));
-    if (key_count_for_prefix_ == 0 || prefix != pre_prefix_.GetKey() ||
+    if (key_count_for_prefix_ == 0 || prefix != pre_prefix_.GetUserKey() ||
         key_count_for_prefix_ % index_sparseness_ == 0) {
       key_count_for_prefix_ = 1;
-      pre_prefix_.SetKey(prefix);
+      pre_prefix_.SetUserKey(prefix);
       size_bytes_pos += EncodeSize(kFullKey, user_key_size, size_bytes);
       Status s = file->Append(Slice(size_bytes, size_bytes_pos));
       if (!s.ok()) {
@@ -128,10 +128,11 @@ Status PlainTableKeyEncoder::AppendKey(const Slice& key,
         // For second key within a prefix, need to encode prefix length
         size_bytes_pos +=
             EncodeSize(kPrefixFromPreviousKey,
-                       static_cast<uint32_t>(pre_prefix_.GetKey().size()),
+                       static_cast<uint32_t>(pre_prefix_.GetUserKey().size()),
                        size_bytes + size_bytes_pos);
       }
-      uint32_t prefix_len = static_cast<uint32_t>(pre_prefix_.GetKey().size());
+      uint32_t prefix_len =
+          static_cast<uint32_t>(pre_prefix_.GetUserKey().size());
       size_bytes_pos += EncodeSize(kKeySuffix, user_key_size - prefix_len,
                                    size_bytes + size_bytes_pos);
       Status s = file->Append(Slice(size_bytes, size_bytes_pos));
@@ -315,9 +316,10 @@ Status PlainTableKeyDecoder::NextPlainEncodingKey(uint32_t start_offset,
   }
   if (!file_reader_.file_info()->is_mmap_mode) {
     cur_key_.SetInternalKey(*parsed_key);
-    parsed_key->user_key = Slice(cur_key_.GetKey().data(), user_key_size);
+    parsed_key->user_key =
+        Slice(cur_key_.GetInternalKey().data(), user_key_size);
     if (internal_key != nullptr) {
-      *internal_key = cur_key_.GetKey();
+      *internal_key = cur_key_.GetInternalKey();
     }
   } else if (internal_key != nullptr) {
     if (decoded_internal_key_valid) {
@@ -325,7 +327,7 @@ Status PlainTableKeyDecoder::NextPlainEncodingKey(uint32_t start_offset,
     } else {
       // Need to copy out the internal key
       cur_key_.SetInternalKey(*parsed_key);
-      *internal_key = cur_key_.GetKey();
+      *internal_key = cur_key_.GetInternalKey();
     }
   }
   return Status::OK();
@@ -369,12 +371,13 @@ Status PlainTableKeyDecoder::NextPrefixEncodingKey(
           // users, because after reading value for the key, the key might
           // be invalid.
           cur_key_.SetInternalKey(*parsed_key);
-          saved_user_key_ = cur_key_.GetKey();
+          saved_user_key_ = cur_key_.GetUserKey();
           if (!file_reader_.file_info()->is_mmap_mode) {
-            parsed_key->user_key = Slice(cur_key_.GetKey().data(), size);
+            parsed_key->user_key =
+                Slice(cur_key_.GetInternalKey().data(), size);
           }
           if (internal_key != nullptr) {
-            *internal_key = cur_key_.GetKey();
+            *internal_key = cur_key_.GetInternalKey();
           }
         } else {
           if (internal_key != nullptr) {
@@ -421,16 +424,16 @@ Status PlainTableKeyDecoder::NextPrefixEncodingKey(
           cur_key_.Reserve(prefix_len_ + size);
           cur_key_.SetInternalKey(tmp, *parsed_key);
           parsed_key->user_key =
-              Slice(cur_key_.GetKey().data(), prefix_len_ + size);
-          saved_user_key_ = cur_key_.GetKey();
+              Slice(cur_key_.GetInternalKey().data(), prefix_len_ + size);
+          saved_user_key_ = cur_key_.GetUserKey();
         } else {
           cur_key_.Reserve(prefix_len_ + size);
           cur_key_.SetInternalKey(Slice(saved_user_key_.data(), prefix_len_),
                                   *parsed_key);
         }
-        parsed_key->user_key = ExtractUserKey(cur_key_.GetKey());
+        parsed_key->user_key = cur_key_.GetUserKey();
         if (internal_key != nullptr) {
-          *internal_key = cur_key_.GetKey();
+          *internal_key = cur_key_.GetInternalKey();
         }
         break;
       }