[yugabyte#11409] DocDB: Skip creating write batch while writing to ro…

…cksdb

Summary:
RocksDB WAL is disabled in our system.
So we could avoid creating write batch while writing to RocksDB.
This diff introduces DirectWriter,that allows us to avoid serializing key and values before writing them to RocksDB.
DirectWriter is used when applying WriteOperation.

In a follow-up diff we could use DirectWriter to also apply transaction intents.

Test Plan: Jenkins

Reviewers: timur, mbautin

Reviewed By: timur, mbautin

Subscribers: mbautin, ybase, bogdan

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

ent/src/yb/master/restore_sys_catalog_state.cc

@@ -26,6 +26,7 @@
 #include "yb/docdb/doc_write_batch.h"
 #include "yb/docdb/docdb.h"
 #include "yb/docdb/pgsql_operation.h"
+#include "yb/docdb/rocksdb_writer.h"
 
 #include "yb/master/catalog_loaders.h"
 #include "yb/master/master_backup.pb.h"
@@ -445,9 +446,9 @@ void RestoreSysCatalogState::WriteToRocksDB(
   docdb::KeyValueWriteBatchPB kv_write_batch;
   write_batch->MoveToWriteBatchPB(&kv_write_batch);
 
+  docdb::NonTransactionalWriter writer(kv_write_batch, write_time);
   rocksdb::WriteBatch rocksdb_write_batch;
-  PrepareNonTransactionWriteBatch(
-      kv_write_batch, write_time, nullptr, &rocksdb_write_batch, nullptr);
+  rocksdb_write_batch.SetDirectWriter(&writer);
   docdb::ConsensusFrontiers frontiers;
   set_op_id(op_id, &frontiers);
   set_hybrid_time(write_time, &frontiers);

src/yb/client/ql-dml-test.cc

@@ -26,6 +26,8 @@
 
 #include "yb/master/master_util.h"
 
+#include "yb/rocksdb/db.h"
+
 #include "yb/tablet/tablet.h"
 #include "yb/tablet/tablet_peer.h"
 

src/yb/docdb/CMakeLists.txt

@@ -28,16 +28,16 @@ ADD_YB_LIBRARY(docdb_proto
         NONLINK_DEPS ${DOCDB_PROTO_TGTS})
 
 set(DOCDB_ENCODING_SRCS
-        doc_key.cc
-        doc_kv_util.cc
-        doc_path.cc
-        key_bounds.cc
-        key_bytes.cc
-        primitive_value.cc
-        primitive_value_util.cc
-        intent.cc
-        doc_scanspec_util.cc
-        )
+    doc_key.cc
+    doc_kv_util.cc
+    doc_path.cc
+    key_bounds.cc
+    key_bytes.cc
+    primitive_value.cc
+    primitive_value_util.cc
+    intent.cc
+    doc_scanspec_util.cc
+    )
 
 set(DOCDB_ENCODING_DEPS
         docdb_proto
@@ -87,6 +87,7 @@ set(DOCDB_SRCS
         ql_rocksdb_storage.cc
         ql_rowwise_iterator_interface.cc
         redis_operation.cc
+        rocksdb_writer.cc
         shared_lock_manager.cc
         subdocument.cc
         subdoc_reader.cc

src/yb/docdb/doc_kv_util.cc

@@ -14,7 +14,7 @@
 #include "yb/docdb/doc_kv_util.h"
 
 #include "yb/docdb/docdb_fwd.h"
-#include "yb/docdb/docdb-internal.h"
+#include "yb/docdb/docdb.h"
 #include "yb/docdb/value_type.h"
 
 #include "yb/util/bytes_formatter.h"
@@ -196,12 +196,9 @@ Result<DocHybridTime> DecodeInvertedDocHt(Slice key_slice) {
         "Invalid doc hybrid time in reverse intent record suffix: $0",
         key_slice.ToDebugHexString());
   }
-  size_t doc_ht_buffer[kMaxWordsPerEncodedHybridTimeWithValueType];
-  memcpy(doc_ht_buffer, key_slice.data(), key_slice.size());
-  for (size_t i = 0; i != kMaxWordsPerEncodedHybridTimeWithValueType; ++i) {
-    doc_ht_buffer[i] = ~doc_ht_buffer[i];
-  }
-  key_slice = Slice(pointer_cast<char*>(doc_ht_buffer), key_slice.size());
+
+  DocHybridTimeWordBuffer doc_ht_buffer;
+  key_slice = InvertEncodedDocHT(key_slice, &doc_ht_buffer);
 
   if (static_cast<ValueType>(key_slice[0]) != ValueType::kHybridTime) {
     return STATUS_FORMAT(
@@ -215,5 +212,13 @@ Result<DocHybridTime> DecodeInvertedDocHt(Slice key_slice) {
   return doc_ht;
 }
 
+Slice InvertEncodedDocHT(const Slice& input, DocHybridTimeWordBuffer* buffer) {
+  memcpy(buffer->data(), input.data(), input.size());
+  for (size_t i = 0; i != kMaxWordsPerEncodedHybridTimeWithValueType; ++i) {
+    (*buffer)[i] = ~(*buffer)[i];
+  }
+  return {pointer_cast<char*>(buffer->data()), input.size()};
+}
+
 }  // namespace docdb
 }  // namespace yb

src/yb/docdb/doc_kv_util.h

@@ -146,6 +146,14 @@ inline std::string ToShortDebugStr(const std::string& raw_str) {
 
 Result<DocHybridTime> DecodeInvertedDocHt(Slice key_slice);
 
+constexpr size_t kMaxWordsPerEncodedHybridTimeWithValueType =
+    ((kMaxBytesPerEncodedHybridTime + 1) + sizeof(size_t) - 1) / sizeof(size_t);
+
+// Puts inverted encoded doc hybrid time specified by input to buffer.
+// And returns slice to it.
+using DocHybridTimeWordBuffer = std::array<size_t, kMaxWordsPerEncodedHybridTimeWithValueType>;
+Slice InvertEncodedDocHT(const Slice& input, DocHybridTimeWordBuffer* buffer);
+
 }  // namespace docdb
 }  // namespace yb
 

src/yb/docdb/docdb-internal.h

@@ -74,9 +74,6 @@ namespace docdb {
 // Infer the key type from the given slice, given whether this is regular or intents RocksDB.
 KeyType GetKeyType(const Slice& slice, StorageDbType db_type);
 
-constexpr size_t kMaxWordsPerEncodedHybridTimeWithValueType =
-    ((kMaxBytesPerEncodedHybridTime + 1) + sizeof(size_t) - 1) / sizeof(size_t);
-
 } // namespace docdb
 } // namespace yb