Polish IterKey and use it in DBImpl::ProcessKeyValueCompaction()

Summary:
1. Polish IterKey a little bit.
2. Turn to use it in local parameter of current_user_key in DBImpl::ProcessKeyValueCompaction(). Our profile showing that DBImpl::ProcessKeyValueCompaction() has about 14% costs in std::string (the base including reading and writing data but excluding compaction filtering), which is higher than it should be. There are two std::string used in DBImpl::ProcessKeyValueCompaction(), compaction_filter_value and current_user_key and it's hard to distinguish the two.

Test Plan: make all check

Reviewers: haobo, ljin

Reviewed By: haobo

CC: igor, yhchiang, dhruba, leveldb

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

Conflicts:
	db/db_impl.cc

db/db_impl.cc

@@ -2540,9 +2540,9 @@ Status DBImpl::ProcessKeyValueCompaction(
   Status status;
   std::string compaction_filter_value;
   ParsedInternalKey ikey;
-  std::string current_user_key;
+  IterKey current_user_key;
   bool has_current_user_key = false;
-  std::vector<char> delete_key;   // for compaction filter
+  IterKey delete_key;
   SequenceNumber last_sequence_for_key __attribute__((unused)) =
     kMaxSequenceNumber;
   SequenceNumber visible_in_snapshot = kMaxSequenceNumber;
@@ -2612,16 +2612,16 @@ Status DBImpl::ProcessKeyValueCompaction(
       // Do not hide error keys
       // TODO: error key stays in db forever? Figure out the intention/rationale
       // v10 error v8 : we cannot hide v8 even though it's pretty obvious.
-      current_user_key.clear();
+      current_user_key.Clear();
       has_current_user_key = false;
       last_sequence_for_key = kMaxSequenceNumber;
       visible_in_snapshot = kMaxSequenceNumber;
     } else {
       if (!has_current_user_key ||
           user_comparator()->Compare(ikey.user_key,
-            Slice(current_user_key)) != 0) {
+            current_user_key.GetKey()) != 0) {
         // First occurrence of this user key
-        current_user_key.assign(ikey.user_key.data(), ikey.user_key.size());
+        current_user_key.SetUserKey(ikey.user_key);
         has_current_user_key = true;
         last_sequence_for_key = kMaxSequenceNumber;
         visible_in_snapshot = kMaxSequenceNumber;
@@ -2642,13 +2642,11 @@ Status DBImpl::ProcessKeyValueCompaction(
                 &compaction_filter_value,
                 &value_changed);
           if (to_delete) {
-            // make a copy of the original key
+            // make a copy of the original key and convert it to a delete
-            delete_key.assign(key.data(), key.data() + key.size());
+            delete_key.SetInternalKey(ExtractUserKey(key), ikey.sequence,
-            // convert it to a delete
+                                      kTypeDeletion);
-            UpdateInternalKey(&delete_key[0], delete_key.size(),
-                ikey.sequence, kTypeDeletion);
             // anchor the key again
-            key = Slice(&delete_key[0], delete_key.size());
+            key = delete_key.GetKey();
             // needed because ikey is backed by key
             ParseInternalKey(key, &ikey);
             // no value associated with delete

db/dbformat.h

@@ -239,47 +239,17 @@ class IterKey {
  public:
   IterKey() : key_(space_), buf_size_(sizeof(space_)), key_size_(0) {}
 
-  ~IterKey() { Clear(); }
+  ~IterKey() { ResetBuffer(); }
 
-  Slice GetKey() const {
+  Slice GetKey() const { return Slice(key_, key_size_); }
-    if (key_ != nullptr) {
-      return Slice(key_, key_size_);
-    } else {
-      return Slice();
-    }
-  }
 
-  bool Valid() const { return key_ != nullptr; }
+  void Clear() { key_size_ = 0; }
-
-  void Clear() {
-    if (key_ != nullptr && key_ != space_) {
-      delete[] key_;
-    }
-    key_ = space_;
-    buf_size_ = sizeof(buf_size_);
-  }
-
-  // Enlarge the buffer size if needed based on key_size.
-  // By default, static allocated buffer is used. Once there is a key
-  // larger than the static allocated buffer, another buffer is dynamically
-  // allocated, until a larger key buffer is requested. In that case, we
-  // reallocate buffer and delete the old one.
-  void EnlargeBufferIfNeeded(size_t key_size) {
-    // If size is smaller than buffer size, continue using current buffer,
-    // or the static allocated one, as default
-    if (key_size > buf_size_) {
-      // Need to enlarge the buffer.
-      Clear();
-      key_ = new char[key_size];
-      buf_size_ = key_size;
-    }
-    key_size_ = key_size;
-  }
 
   void SetUserKey(const Slice& user_key) {
     size_t size = user_key.size();
     EnlargeBufferIfNeeded(size);
     memcpy(key_, user_key.data(), size);
+    key_size_ = size;
   }
 
   void SetInternalKey(const Slice& user_key, SequenceNumber s,
@@ -288,6 +258,7 @@ class IterKey {
     EnlargeBufferIfNeeded(usize + sizeof(uint64_t));
     memcpy(key_, user_key.data(), usize);
     EncodeFixed64(key_ + usize, PackSequenceAndType(s, value_type));
+    key_size_ = usize + sizeof(uint64_t);
   }
 
   void SetInternalKey(const ParsedInternalKey& parsed_key) {
@@ -300,6 +271,31 @@ class IterKey {
   size_t key_size_;
   char space_[32];  // Avoid allocation for short keys
 
+  void ResetBuffer() {
+    if (key_ != nullptr && key_ != space_) {
+      delete[] key_;
+    }
+    key_ = space_;
+    buf_size_ = sizeof(buf_size_);
+    key_size_ = 0;
+  }
+
+  // Enlarge the buffer size if needed based on key_size.
+  // By default, static allocated buffer is used. Once there is a key
+  // larger than the static allocated buffer, another buffer is dynamically
+  // allocated, until a larger key buffer is requested. In that case, we
+  // reallocate buffer and delete the old one.
+  void EnlargeBufferIfNeeded(size_t key_size) {
+    // If size is smaller than buffer size, continue using current buffer,
+    // or the static allocated one, as default
+    if (key_size > buf_size_) {
+      // Need to enlarge the buffer.
+      ResetBuffer();
+      key_ = new char[key_size];
+      buf_size_ = key_size;
+    }
+  }
+
   // No copying allowed
   IterKey(const IterKey&) = delete;
   void operator=(const IterKey&) = delete;