Enable moving a string or PinnableSlice into PinnableWideColumns (#11248)

Summary:
This makes it possible to eliminate some copies in `GetEntity` / `MultiGetEntity`,
in particular when `Merge`s or blobs are involved.

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

Test Plan: `make check`

Reviewed By: akankshamahajan15

Differential Revision: D43544215

Pulled By: ltamasi

fbshipit-source-id: bc4c8955a24bbd8bc4ab098e72133ead757f9707

db/memtable.cc

@@ -1078,7 +1078,7 @@ static bool SaveValue(void* arg, const char* entry) {
                 *(s->value) = std::move(result);
               } else {
                 assert(s->columns);
-                s->columns->SetPlainValue(result);
+                s->columns->SetPlainValue(std::move(result));
               }
             }
           }
@@ -1152,7 +1152,7 @@ static bool SaveValue(void* arg, const char* entry) {
                 /* op_failure_scope */ nullptr);
 
             if (s->status->ok()) {
-              *(s->status) = s->columns->SetWideColumnValue(result);
+              *(s->status) = s->columns->SetWideColumnValue(std::move(result));
             }
           }
         } else if (s->value) {
@@ -1200,7 +1200,7 @@ static bool SaveValue(void* arg, const char* entry) {
                 *(s->value) = std::move(result);
               } else {
                 assert(s->columns);
-                s->columns->SetPlainValue(result);
+                s->columns->SetPlainValue(std::move(result));
               }
             }
           } else {
@@ -1249,7 +1249,7 @@ static bool SaveValue(void* arg, const char* entry) {
                 *(s->value) = std::move(result);
               } else {
                 assert(s->columns);
-                s->columns->SetPlainValue(result);
+                s->columns->SetPlainValue(std::move(result));
               }
             }
           }

db/version_set.cc

@@ -2233,7 +2233,7 @@ void Version::MultiGetBlob(
           range.AddValueSize(key_context->value->size());
         } else {
           assert(key_context->columns);
-          key_context->columns->SetPlainValue(blob.result);
+          key_context->columns->SetPlainValue(std::move(blob.result));
           range.AddValueSize(key_context->columns->serialized_size());
         }
 
@@ -2390,8 +2390,7 @@ void Version::Get(const ReadOptions& read_options, const LookupKey& k,
             *value = std::move(result);
           } else {
             assert(columns);
-            columns->Reset();
+            columns->SetPlainValue(std::move(result));
-            columns->SetPlainValue(result);
           }
         }
 
@@ -2445,7 +2444,7 @@ void Version::Get(const ReadOptions& read_options, const LookupKey& k,
           value->PinSelf();
         } else {
           assert(columns != nullptr);
-          columns->SetPlainValue(result);
+          columns->SetPlainValue(std::move(result));
         }
       }
     }
@@ -2696,7 +2695,7 @@ void Version::MultiGet(const ReadOptions& read_options, MultiGetRange* range,
         range->AddValueSize(iter->value->size());
       } else {
         assert(iter->columns);
-        iter->columns->SetPlainValue(result);
+        iter->columns->SetPlainValue(std::move(result));
         range->AddValueSize(iter->columns->serialized_size());
       }
 

include/rocksdb/wide_columns.h

@@ -97,15 +97,22 @@ class PinnableWideColumns {
 
   void SetPlainValue(const Slice& value);
   void SetPlainValue(const Slice& value, Cleanable* cleanable);
+  void SetPlainValue(PinnableSlice&& value);
+  void SetPlainValue(std::string&& value);
 
   Status SetWideColumnValue(const Slice& value);
   Status SetWideColumnValue(const Slice& value, Cleanable* cleanable);
+  Status SetWideColumnValue(PinnableSlice&& value);
+  Status SetWideColumnValue(std::string&& value);
 
   void Reset();
 
  private:
   void CopyValue(const Slice& value);
   void PinOrCopyValue(const Slice& value, Cleanable* cleanable);
+  void MoveValue(PinnableSlice&& value);
+  void MoveValue(std::string&& value);
+
   void CreateIndexForPlainValue();
   Status CreateIndexForWideColumns();
 
@@ -127,6 +134,18 @@ inline void PinnableWideColumns::PinOrCopyValue(const Slice& value,
   value_.PinSlice(value, cleanable);
 }
 
+inline void PinnableWideColumns::MoveValue(PinnableSlice&& value) {
+  value_ = std::move(value);
+}
+
+inline void PinnableWideColumns::MoveValue(std::string&& value) {
+  std::string* const buf = value_.GetSelf();
+  assert(buf);
+
+  *buf = std::move(value);
+  value_.PinSelf();
+}
+
 inline void PinnableWideColumns::CreateIndexForPlainValue() {
   columns_ = WideColumns{{kDefaultWideColumnName, value_}};
 }
@@ -142,6 +161,16 @@ inline void PinnableWideColumns::SetPlainValue(const Slice& value,
   CreateIndexForPlainValue();
 }
 
+inline void PinnableWideColumns::SetPlainValue(PinnableSlice&& value) {
+  MoveValue(std::move(value));
+  CreateIndexForPlainValue();
+}
+
+inline void PinnableWideColumns::SetPlainValue(std::string&& value) {
+  MoveValue(std::move(value));
+  CreateIndexForPlainValue();
+}
+
 inline Status PinnableWideColumns::SetWideColumnValue(const Slice& value) {
   CopyValue(value);
   return CreateIndexForWideColumns();
@@ -153,6 +182,16 @@ inline Status PinnableWideColumns::SetWideColumnValue(const Slice& value,
   return CreateIndexForWideColumns();
 }
 
+inline Status PinnableWideColumns::SetWideColumnValue(PinnableSlice&& value) {
+  MoveValue(std::move(value));
+  return CreateIndexForWideColumns();
+}
+
+inline Status PinnableWideColumns::SetWideColumnValue(std::string&& value) {
+  MoveValue(std::move(value));
+  return CreateIndexForWideColumns();
+}
+
 inline void PinnableWideColumns::Reset() {
   value_.Reset();
   columns_.clear();

table/get_context.cc

@@ -489,7 +489,7 @@ void GetContext::Merge(const Slice* value) {
   }
 
   assert(columns_);
-  columns_->SetPlainValue(result);
+  columns_->SetPlainValue(std::move(result));
 }
 
 void GetContext::MergeWithEntity(Slice entity) {
@@ -552,7 +552,7 @@ void GetContext::MergeWithEntity(Slice entity) {
 
   {
     assert(columns_);
-    const Status s = columns_->SetWideColumnValue(result);
+    const Status s = columns_->SetWideColumnValue(std::move(result));
     if (!s.ok()) {
       state_ = kCorrupt;
       return;