Support providing the default column separately when serializing columns (#10839)

Summary:
The patch makes it possible to provide the value of the default column
separately when calling `WideColumnSerialization::Serialize`. This eliminates
the need to construct a new `WideColumns` vector in certain cases
(for example, it will come in handy when implementing `Merge`).

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

Test Plan: `make check`

Reviewed By: riversand963

Differential Revision: D40561448

Pulled By: ltamasi

fbshipit-source-id: 69becdd510e6a83ab1feb956c12772110e1040d6

db/wide/wide_column_serialization.cc

@@ -15,16 +15,32 @@
 
 namespace ROCKSDB_NAMESPACE {
 
-Status WideColumnSerialization::Serialize(const WideColumns& columns,
+Status WideColumnSerialization::SerializeImpl(const Slice* value_of_default,
+                                              const WideColumns& columns,
                                               std::string& output) {
-  if (columns.size() >
-      static_cast<size_t>(std::numeric_limits<uint32_t>::max())) {
+  const size_t num_columns =
+      value_of_default ? columns.size() + 1 : columns.size();
+
+  if (num_columns > static_cast<size_t>(std::numeric_limits<uint32_t>::max())) {
     return Status::InvalidArgument("Too many wide columns");
   }
 
   PutVarint32(&output, kCurrentVersion);
 
-  PutVarint32(&output, static_cast<uint32_t>(columns.size()));
+  PutVarint32(&output, static_cast<uint32_t>(num_columns));
+
+  const Slice* prev_name = nullptr;
+  if (value_of_default) {
+    if (value_of_default->size() >
+        static_cast<size_t>(std::numeric_limits<uint32_t>::max())) {
+      return Status::InvalidArgument("Wide column value too long");
+    }
+
+    PutLengthPrefixedSlice(&output, kDefaultWideColumnName);
+    PutVarint32(&output, static_cast<uint32_t>(value_of_default->size()));
+
+    prev_name = &kDefaultWideColumnName;
+  }
 
   for (size_t i = 0; i < columns.size(); ++i) {
     const WideColumn& column = columns[i];
@@ -34,7 +50,8 @@ Status WideColumnSerialization::Serialize(const WideColumns& columns,
         static_cast<size_t>(std::numeric_limits<uint32_t>::max())) {
       return Status::InvalidArgument("Wide column name too long");
     }
-    if (i > 0 && columns[i - 1].name().compare(name) >= 0) {
+
+    if (prev_name && prev_name->compare(name) >= 0) {
       return Status::Corruption("Wide columns out of order");
     }
 
@@ -46,6 +63,12 @@ Status WideColumnSerialization::Serialize(const WideColumns& columns,
 
     PutLengthPrefixedSlice(&output, name);
     PutVarint32(&output, static_cast<uint32_t>(value.size()));
+
+    prev_name = &name;
+  }
+
+  if (value_of_default) {
+    output.append(value_of_default->data(), value_of_default->size());
   }
 
   for (const auto& column : columns) {

db/wide/wide_column_serialization.h

@@ -44,6 +44,10 @@ class Slice;
 class WideColumnSerialization {
  public:
   static Status Serialize(const WideColumns& columns, std::string& output);
+  static Status Serialize(const Slice& value_of_default,
+                          const WideColumns& other_columns,
+                          std::string& output);
+
   static Status Deserialize(Slice& input, WideColumns& columns);
 
   static WideColumns::const_iterator Find(const WideColumns& columns,
@@ -51,6 +55,23 @@ class WideColumnSerialization {
   static Status GetValueOfDefaultColumn(Slice& input, Slice& value);
 
   static constexpr uint32_t kCurrentVersion = 1;
+
+ private:
+  static Status SerializeImpl(const Slice* value_of_default,
+                              const WideColumns& columns, std::string& output);
 };
 
+inline Status WideColumnSerialization::Serialize(const WideColumns& columns,
+                                                 std::string& output) {
+  constexpr Slice* value_of_default = nullptr;
+
+  return SerializeImpl(value_of_default, columns, output);
+}
+
+inline Status WideColumnSerialization::Serialize(
+    const Slice& value_of_default, const WideColumns& other_columns,
+    std::string& output) {
+  return SerializeImpl(&value_of_default, other_columns, output);
+}
+
 }  // namespace ROCKSDB_NAMESPACE

db/wide/wide_column_serialization_test.cc

@@ -124,6 +124,25 @@ TEST(WideColumnSerializationTest, SerializeDeserialize) {
   }
 }
 
+TEST(WideColumnSerializationTest, SerializeWithPrepend) {
+  Slice value_of_default("baz");
+  WideColumns other_columns{{"foo", "bar"}, {"hello", "world"}};
+
+  std::string output;
+  ASSERT_OK(WideColumnSerialization::Serialize(value_of_default, other_columns,
+                                               output));
+
+  Slice input(output);
+
+  WideColumns deserialized_columns;
+  ASSERT_OK(WideColumnSerialization::Deserialize(input, deserialized_columns));
+
+  WideColumns expected_columns{{kDefaultWideColumnName, value_of_default},
+                               other_columns[0],
+                               other_columns[1]};
+  ASSERT_EQ(deserialized_columns, expected_columns);
+}
+
 TEST(WideColumnSerializationTest, SerializeDuplicateError) {
   WideColumns columns{{"foo", "bar"}, {"foo", "baz"}};
   std::string output;
@@ -132,6 +151,16 @@ TEST(WideColumnSerializationTest, SerializeDuplicateError) {
       WideColumnSerialization::Serialize(columns, output).IsCorruption());
 }
 
+TEST(WideColumnSerializationTest, SerializeWithPrependDuplicateError) {
+  Slice value_of_default("baz");
+  WideColumns other_columns{{kDefaultWideColumnName, "dup"}, {"foo", "bar"}};
+
+  std::string output;
+  ASSERT_TRUE(WideColumnSerialization::Serialize(value_of_default,
+                                                 other_columns, output)
+                  .IsCorruption());
+}
+
 TEST(WideColumnSerializationTest, SerializeOutOfOrderError) {
   WideColumns columns{{"hello", "world"}, {"foo", "bar"}};
   std::string output;