Implement log blobs

Summary:
This patch adds the ability for the user to add sequences of arbitrary data (blobs) to write batches. These blobs are saved to the log along with everything else in the write batch. You can add multiple blobs per WriteBatch and the ordering of blobs, puts, merges, and deletes are preserved.

Blobs are not saves to SST files. RocksDB ignores blobs in every way except for writing them to the log.

Before committing this patch, I need to add some test code. But I'm submitting it now so people can comment on the API.

Test Plan: make -j32 check

Reviewers: dhruba, haobo, vamsi

Reviewed By: dhruba

CC: leveldb

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

db/db_iter.cc

@@ -217,6 +217,9 @@ void DBIter::FindNextUserEntry(bool skipping) {
             // TODO: what if !iter_->Valid()
             return;
             break;
+          case kTypeLogData:
+            assert(false);
+            break;
         }
       }
     }

db/db_test.cc

@@ -457,6 +457,9 @@ class DBTest {
             case kTypeDeletion:
               result += "DEL";
               break;
+            case kTypeLogData:
+              assert(false);
+              break;
           }
         }
         iter->Next();
@@ -3132,6 +3135,48 @@ TEST(DBTest, TransactionLogIteratorBatchOperations) {
   } while (ChangeCompactOptions());
 }
 
+TEST(DBTest, TransactionLogIteratorBlobs) {
+  Options options = OptionsForLogIterTest();
+  DestroyAndReopen(&options);
+  {
+    WriteBatch batch;
+    batch.Put("key1", DummyString(1024));
+    batch.Put("key2", DummyString(1024));
+    batch.PutLogData(Slice("blob1"));
+    batch.Put("key3", DummyString(1024));
+    batch.PutLogData(Slice("blob2"));
+    batch.Delete("key2");
+    dbfull()->Write(WriteOptions(), &batch);
+    Reopen(&options);
+  }
+
+  auto res = OpenTransactionLogIter(0)->GetBatch();
+  struct Handler : public WriteBatch::Handler {
+    std::string seen;
+    virtual void Put(const Slice& key, const Slice& value) {
+      seen += "Put(" + key.ToString() + ", " + std::to_string(value.size()) +
+        ")";
+    }
+    virtual void Merge(const Slice& key, const Slice& value) {
+      seen += "Merge(" + key.ToString() + ", " + std::to_string(value.size()) +
+        ")";
+    }
+    virtual void LogData(const Slice& blob) {
+      seen += "LogData(" + blob.ToString() + ")";
+    }
+    virtual void Delete(const Slice& key) {
+      seen += "Delete(" + key.ToString() + ")";
+    }
+  } handler;
+  res.writeBatchPtr->Iterate(&handler);
+  ASSERT_EQ("Put(key1, 1024)"
+            "Put(key2, 1024)"
+            "LogData(blob1)"
+            "Put(key3, 1024)"
+            "LogData(blob2)"
+            "Delete(key2)", handler.seen);
+}
+
 TEST(DBTest, ReadCompaction) {
   std::string value(4096, '4'); // a string of size 4K
   {

db/dbformat.h

@@ -25,7 +25,8 @@ class InternalKey;
 enum ValueType {
   kTypeDeletion = 0x0,
   kTypeValue = 0x1,
-  kTypeMerge = 0x2
+  kTypeMerge = 0x2,
+  kTypeLogData = 0x3
 };
 // kValueTypeForSeek defines the ValueType that should be passed when
 // constructing a ParsedInternalKey object for seeking to a particular

db/memtable.cc

@@ -211,6 +211,9 @@ bool MemTable::Get(const LookupKey& key, std::string* value, Status* s,
           }
           break;
         }
+        case kTypeLogData:
+          assert(false);
+          break;
       }
     } else {
       // exit loop if user key does not match

db/version_set.cc

@@ -330,6 +330,10 @@ static bool SaveValue(void* arg, const Slice& ikey, const Slice& v, bool didIO){
             }
           }
           return true;
+
+        case kTypeLogData:
+          assert(false);
+          break;
       }
     }
   }

db/write_batch.cc

@@ -43,6 +43,11 @@ void WriteBatch::Handler::Merge(const Slice& key, const Slice& value) {
   throw std::runtime_error("Handler::Merge not implemented!");
 }
 
+void WriteBatch::Handler::LogData(const Slice& blob) {
+  // If the user has not specified something to do with blobs, then we ignore
+  // them.
+}
+
 void WriteBatch::Clear() {
   rep_.clear();
   rep_.resize(kHeader);
@@ -59,10 +64,9 @@ Status WriteBatch::Iterate(Handler* handler) const {
   }
 
   input.remove_prefix(kHeader);
-  Slice key, value;
+  Slice key, value, blob;
   int found = 0;
   while (!input.empty()) {
-    found++;
     char tag = input[0];
     input.remove_prefix(1);
     switch (tag) {
@@ -70,6 +74,7 @@ Status WriteBatch::Iterate(Handler* handler) const {
         if (GetLengthPrefixedSlice(&input, &key) &&
             GetLengthPrefixedSlice(&input, &value)) {
           handler->Put(key, value);
+          found++;
         } else {
           return Status::Corruption("bad WriteBatch Put");
         }
@@ -77,6 +82,7 @@ Status WriteBatch::Iterate(Handler* handler) const {
       case kTypeDeletion:
         if (GetLengthPrefixedSlice(&input, &key)) {
           handler->Delete(key);
+          found++;
         } else {
           return Status::Corruption("bad WriteBatch Delete");
         }
@@ -85,10 +91,18 @@ Status WriteBatch::Iterate(Handler* handler) const {
         if (GetLengthPrefixedSlice(&input, &key) &&
             GetLengthPrefixedSlice(&input, &value)) {
           handler->Merge(key, value);
+          found++;
         } else {
           return Status::Corruption("bad WriteBatch Merge");
         }
         break;
+      case kTypeLogData:
+        if (GetLengthPrefixedSlice(&input, &blob)) {
+          handler->LogData(blob);
+        } else {
+          return Status::Corruption("bad WriteBatch Blob");
+        }
+        break;
       default:
         return Status::Corruption("unknown WriteBatch tag");
     }
@@ -136,6 +150,10 @@ void WriteBatch::Merge(const Slice& key, const Slice& value) {
   PutLengthPrefixedSlice(&rep_, value);
 }
 
+void WriteBatch::PutLogData(const Slice& blob) {
+  rep_.push_back(static_cast<char>(kTypeLogData));
+  PutLengthPrefixedSlice(&rep_, blob);
+}
 
 namespace {
 class MemTableInserter : public WriteBatch::Handler {

db/write_batch_test.cc

@@ -50,13 +50,16 @@ static std::string PrintContents(WriteBatch* b) {
         state.append(")");
         count++;
         break;
+      case kTypeLogData:
+        assert(false);
+        break;
     }
     state.append("@");
     state.append(NumberToString(ikey.sequence));
   }
   delete iter;
   if (!s.ok()) {
-    state.append("ParseError()");
+    state.append(s.ToString());
   } else if (count != WriteBatchInternal::Count(b)) {
     state.append("CountMismatch()");
   }
@@ -97,7 +100,7 @@ TEST(WriteBatchTest, Corruption) {
   WriteBatchInternal::SetContents(&batch,
                                   Slice(contents.data(),contents.size()-1));
   ASSERT_EQ("Put(foo, bar)@200"
-            "ParseError()",
+            "Corruption: bad WriteBatch Delete",
             PrintContents(&batch));
 }
 
@@ -131,6 +134,50 @@ TEST(WriteBatchTest, Append) {
   ASSERT_EQ(4, b1.Count());
 }
 
+TEST(WriteBatchTest, Blob) {
+  WriteBatch batch;
+  batch.Put(Slice("k1"), Slice("v1"));
+  batch.Put(Slice("k2"), Slice("v2"));
+  batch.Put(Slice("k3"), Slice("v3"));
+  batch.PutLogData(Slice("blob1"));
+  batch.Delete(Slice("k2"));
+  batch.PutLogData(Slice("blob2"));
+  batch.Merge(Slice("foo"), Slice("bar"));
+  ASSERT_EQ(5, batch.Count());
+  ASSERT_EQ("Merge(foo, bar)@4"
+            "Put(k1, v1)@0"
+            "Delete(k2)@3"
+            "Put(k2, v2)@1"
+            "Put(k3, v3)@2",
+            PrintContents(&batch));
+
+  struct Handler : public WriteBatch::Handler {
+    std::string seen;
+    virtual void Put(const Slice& key, const Slice& value) {
+      seen += "Put(" + key.ToString() + ", " + value.ToString() + ")";
+    }
+    virtual void Merge(const Slice& key, const Slice& value) {
+      seen += "Merge(" + key.ToString() + ", " + value.ToString() + ")";
+    }
+    virtual void LogData(const Slice& blob) {
+      seen += "LogData(" + blob.ToString() + ")";
+    }
+    virtual void Delete(const Slice& key) {
+      seen += "Delete(" + key.ToString() + ")";
+    }
+  } handler;
+  batch.Iterate(&handler);
+  ASSERT_EQ(
+            "Put(k1, v1)"
+            "Put(k2, v2)"
+            "Put(k3, v3)"
+            "LogData(blob1)"
+            "Delete(k2)"
+            "LogData(blob2)"
+            "Merge(foo, bar)",
+            handler.seen);
+}
+
 }  // namespace leveldb
 
 int main(int argc, char** argv) {