#2298 VERIFY_RESULT creates undesirable copy of Result<T&>'s value

Summary:
1. `VERIFY_RESULT` and similar macros uses GNU expression statement `__extension__`

```
#define VERIFY_RESULT(expr) \
  __extension__ ({ auto&& __result = (expr); RETURN_NOT_OK(__result); std::move(*__result); })
```

In spite of the fact last statement is a rvalue reference copy object is created by using of `T(T&&)` move-constructor (if any) for `Result<T>/Result<T&>` situation and `T::T(const T&)` copy-constructor (if any) for `Result<const T&>`

To avoid undesirable copying of `T` object in case of `Result<T&>` and `Result<const T&>` `std::reference_wrapper<T>` is returned from expression statement `__extension__`.
As of now `VERIFY_RESULT` and similar macros returns `std::reference_wrapper<T>` for `Result<T&>/Result<const T&>` situation new helper macro `VERIFY_RESULT_REF` is introduced (it is still safe to use `VERIFY_RESULT` in this situation).

It is useful with `auto`:
```
Result<const vector<int>&> GetVector() {
  static const vector<int> values{1, 2, 3};
  return values;
}

const auto& value = VERIFY_RESULT_REF(GetVector());
...
for(const auto& i : VERIFY_RESULT_REF(GetVector())) {
...
}
```

2. Make `Status::OK()` be a wrapper to avoid creation of `Result<T>` from it at compile time

Test Plan:
New unit test was introduced
```
./yb_build.sh --cxx-test result-test
```

Reviewers: mikhail, sergei

Reviewed By: sergei

Subscribers: bogdan, ybase

Differential Revision: https://phabricator.dev.yugabyte.com/D7204

ent/src/yb/cdc/cdc_producer.cc

@@ -255,7 +255,7 @@ Status CDCProducer::PopulateWriteRecord(const ReplicateMsgPtr& msg,
   CDCRecordPB* record = nullptr;
   for (const auto& write_pair : batch.write_pairs()) {
     Slice key = write_pair.key();
-    const auto& key_sizes = VERIFY_RESULT(docdb::DocKey::EncodedHashPartAndDocKeySizes(key));
+    const auto key_sizes = VERIFY_RESULT(docdb::DocKey::EncodedHashPartAndDocKeySizes(key));
 
     Slice value = write_pair.value();
     docdb::Value decoded_value;
@@ -309,7 +309,7 @@ Status CDCProducer::PopulateWriteRecord(const ReplicateMsgPtr& msg,
       Slice key_column = write_pair.key().data() + key_sizes.second;
       RETURN_NOT_OK(PrimitiveValue::DecodeKey(&key_column, &column_id));
       if (column_id.value_type() == docdb::ValueType::kColumnId) {
-        ColumnSchema col = VERIFY_RESULT(schema.column_by_id(column_id.GetColumnId()));
+        const ColumnSchema& col = VERIFY_RESULT(schema.column_by_id(column_id.GetColumnId()));
         AddColumnToMap(col, decoded_value.primitive_value(), record->add_changes());
       } else if (column_id.value_type() != docdb::ValueType::kSystemColumnId) {
         LOG(DFATAL) << "Unexpected value type in key: " << column_id.value_type();

src/yb/client/client.cc

@@ -990,7 +990,7 @@ Result<bool> YBClient::IsLoadBalanced(uint32_t num_servers) {
   req.set_expected_num_servers(num_servers);
   // Cannot use CALL_SYNC_LEADER_MASTER_RPC directly since this is susbstituted with RETURN_NOT_OK
   // and we want to capture the status to check if load is balanced.
-  Status s = [&, this]() {
+  Status s = [&, this]() -> Status {
     CALL_SYNC_LEADER_MASTER_RPC(req, resp, IsLoadBalanced);
     return Status::OK();
   }();

src/yb/common/ql_type.h

@@ -280,7 +280,7 @@ class QLType {
   }
 
   // Returns the type of given field, or nullptr if that field is not found in this UDT.R
-  const Result<QLType::SharedPtr> GetUDTFieldTypeByName(const std::string& field_name) const {
+  Result<QLType::SharedPtr> GetUDTFieldTypeByName(const std::string& field_name) const {
     SCHECK(IsUserDefined(), InternalError, "Can only be called on UDT");
     const int idx = GetUDTypeFieldIdxByName(field_name);
     if (idx == -1) {

src/yb/consensus/replica_state.cc

@@ -800,7 +800,7 @@ Result<bool> ReplicaState::AdvanceCommittedOpIdUnlocked(
     VLOG_WITH_PREFIX(1)
         << "Already marked ops through " << last_committed_op_id_ << " as committed. "
         << "Now trying to mark " << committed_op_id << " which would be a no-op.";
-    return Status::OK();
+    return false;
   }
 
   if (pending_operations_.empty()) {

src/yb/docdb/cql_operation.cc

@@ -813,7 +813,7 @@ Status QLWriteOperation::Apply(const DocOperationApplyData& data) {
           CHECK(column_value.has_column_id())
               << "column id missing: " << column_value.DebugString();
           const ColumnId column_id(column_value.column_id());
-          const auto column = VERIFY_RESULT(schema_.column_by_id(column_id));
+          const auto& column = VERIFY_RESULT_REF(schema_.column_by_id(column_id));
           const DocPath sub_path(
               column.is_static() ?
                 encoded_hashed_doc_key_.as_slice() : encoded_pk_doc_key_.as_slice(),

src/yb/docdb/doc_key.cc

@@ -80,7 +80,7 @@ Status ConsumePrimitiveValuesFromKey(Slice* slice, AllowSpecial allow_special, C
 
 Status ConsumePrimitiveValuesFromKey(Slice* slice, AllowSpecial allow_special,
                                      boost::container::small_vector_base<Slice>* result) {
-  return ConsumePrimitiveValuesFromKey(slice, allow_special, [slice, result] {
+  return ConsumePrimitiveValuesFromKey(slice, allow_special, [slice, result]() -> Status {
     auto begin = slice->data();
     RETURN_NOT_OK(PrimitiveValue::DecodeKey(slice, /* out */ nullptr));
     if (result) {

src/yb/docdb/doc_rowwise_iterator.cc

@@ -217,7 +217,7 @@ Status DiscreteScanChoices::DoneWithCurrentTarget() {
 Status DiscreteScanChoices::SkipTargetsUpTo(const Slice& new_target) {
   VLOG(2) << __PRETTY_FUNCTION__ << " Updating current target to be >= " << new_target;
   DCHECK(!FinishedWithScanChoices());
-  InitScanTargetRangeGroupIfNeeded();
+  RETURN_NOT_OK(InitScanTargetRangeGroupIfNeeded());
   DocKeyDecoder decoder(new_target);
   RETURN_NOT_OK(decoder.DecodeToRangeGroup());
   current_scan_target_.Reset(Slice(new_target.data(), decoder.left_input().data()));

src/yb/rocksdb/db/db_impl.cc

@@ -3309,7 +3309,7 @@ Result<FileNumbersHolder> DBImpl::BackgroundCompaction(
       delete cfd;
       // This was the last reference of the column family, so no need to
       // compact.
-      return Status::OK();
+      return FileNumbersHolder();
     }
 
     if (is_large_compaction) {

src/yb/rpc/reactor.cc

@@ -289,7 +289,7 @@ void Reactor::QueueEventOnAllConnections(
 
 Status Reactor::DumpRunningRpcs(const DumpRunningRpcsRequestPB& req,
                                 DumpRunningRpcsResponsePB* resp) {
-  return RunOnReactorThread([&req, resp](Reactor* reactor) {
+  return RunOnReactorThread([&req, resp](Reactor* reactor) -> Status {
     for (const ConnectionPtr& conn : reactor->server_conns_) {
       RETURN_NOT_OK(conn->DumpPB(req, resp->add_inbound_connections()));
     }

src/yb/rpc/tcp_stream.cc

@@ -241,7 +241,7 @@ void TcpStream::PopSending() {
 
 void TcpStream::Handler(ev::io& watcher, int revents) {  // NOLINT
   DVLOG_WITH_PREFIX(4) << "Handler(revents=" << revents << ")";
-  auto status = Status::OK();
+  Status status = Status::OK();
   if (revents & ev::ERROR) {
     status = STATUS(NetworkError, ToString() + ": Handler encountered an error");
     VLOG_WITH_PREFIX(3) << status;

src/yb/tserver/mini_tablet_server.cc

@@ -194,7 +194,7 @@ Status MiniTabletServer::FlushTablets(tablet::FlushMode mode, tablet::FlushFlags
   if (!server_) {
     return Status::OK();
   }
-  return ForAllTablets(this, [mode, flags](TabletPeer* tablet_peer) {
+  return ForAllTablets(this, [mode, flags](TabletPeer* tablet_peer) -> Status {
     if (!tablet_peer->tablet()) {
       return Status::OK();
     }

src/yb/tserver/tablet_service.cc

@@ -1473,7 +1473,7 @@ Result<ReadHybridTime> TabletServiceImpl::DoRead(ReadContext* read_context) {
     return STATUS(NotSupported, "Transaction status table does not support read");
   }
 
-  return Status::OK();
+  return ReadHybridTime();
 }
 
 ConsensusServiceImpl::ConsensusServiceImpl(const scoped_refptr<MetricEntity>& metric_entity,

src/yb/util/fast_varint.cc

@@ -352,13 +352,10 @@ Result<uint64_t> FastDecodeUnsignedVarInt(Slice* slice) {
 
 Result<uint64_t> FastDecodeUnsignedVarInt(const Slice& slice) {
   Slice s(slice);
-  auto status = FastDecodeUnsignedVarInt(&s);
-  if (!status.ok()) {
-    return status;
-  }
+  const auto result = VERIFY_RESULT(FastDecodeUnsignedVarInt(&s));
   if (s.size() != 0)
     return STATUS(Corruption, "Slice not fully decoded.");
-  return Status::OK();
+  return result;
 }
 
 }  // namespace util

src/yb/util/pb_util.cc

@@ -583,7 +583,7 @@ Status ReadablePBContainerFile::Dump(ostream* os, bool oneline) {
       count++;
     }
   }
-  return s.IsEndOfFile() ? s.OK() : s;
+  return s.IsEndOfFile() ? Status::OK() : s;
 }
 
 Status ReadablePBContainerFile::Close() {

src/yb/util/result-test.cc

@@ -226,5 +226,58 @@ TEST_F(ResultTest, VerifyResultMacro) {
   ASSERT_TRUE(s.IsInternalError());
 }
 
+struct NonCopyableNonMovable {
+  NonCopyableNonMovable() = default;
+  NonCopyableNonMovable(const NonCopyableNonMovable&) = delete;
+  NonCopyableNonMovable& operator=(const NonCopyableNonMovable&) = delete;
+};
+
+class MoveCounter {
+ public:
+  MoveCounter() = default;
+  MoveCounter(const MoveCounter&) = delete;
+
+  MoveCounter(const MoveCounter&&) {
+    ++counter_;
+  }
+
+  MoveCounter& operator=(const MoveCounter&) = delete;
+
+  MoveCounter& operator=(MoveCounter&&) {
+    ++counter_;
+    return *this;
+  }
+
+  static size_t counter() {
+    return counter_;
+  }
+
+ private:
+  static size_t counter_;
+};
+
+size_t MoveCounter::counter_ = 0;
+
+// Next function won't compile in case VERIFY_RESULT will try to create copy of Result's data
+Status VerifyResultMacroReferenceNoCopyHelper() {
+  static const NonCopyableNonMovable data;
+  const auto& r ATTRIBUTE_UNUSED = VERIFY_RESULT_REF(Result<const NonCopyableNonMovable&>(data));
+  return Status::OK();
+}
+
+Status VerifyResultMacroMoveCountHelper() {
+  const auto result ATTRIBUTE_UNUSED = VERIFY_RESULT(Result<MoveCounter>(MoveCounter()));
+  return Status::OK();
+}
+
+TEST_F(ResultTest, VerifyResultMacroReferenceNoCopy) {
+  VerifyResultMacroReferenceNoCopyHelper();
+}
+
+TEST_F(ResultTest, VerifyResultMacroMoveCount) {
+  VerifyResultMacroMoveCountHelper();
+  ASSERT_EQ(2, MoveCounter::counter());
+}
+
 } // namespace test
 } // namespace yb