Split BlobFileState into an immutable and a mutable part (#6502)

Summary:
It's never too soon to refactor something. The patch splits the recently
introduced (`VersionEdit` related) `BlobFileState` into two classes
`BlobFileAddition` and `BlobFileGarbage`. The idea is that once blob files
are closed, they are immutable, and the only thing that changes is the
amount of garbage in them. In the new design, `BlobFileAddition` contains
the immutable attributes (currently, the count and total size of all blobs, checksum
method, and checksum value), while `BlobFileGarbage` contains the mutable
GC-related information elements (count and total size of garbage blobs). This is a
better fit for the GC logic and is more consistent with how SST files are handled.
Pull Request resolved: https://github.com/facebook/rocksdb/pull/6502

Test Plan: `make check`

Differential Revision: D20348352

Pulled By: ltamasi

fbshipit-source-id: ff93f0121e80ab15e0e0a6525ba0d6af16a0e008

16 files changed, 797 insertions, 472 deletions

diff --git a/CMakeLists.txt b/CMakeLists.txt
index 7ec3e7d23..18413f173 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -501,7 +501,8 @@ set(SOURCES
         cache/lru_cache.cc
         cache/sharded_cache.cc
         db/arena_wrapped_db_iter.cc
-        db/blob_file_state.cc
+        db/blob_file_addition.cc
+        db/blob_file_garbage.cc
         db/builder.cc
         db/c.cc
         db/column_family.cc
@@ -925,7 +926,8 @@ if(WITH_TESTS)
   set(TESTS
         cache/cache_test.cc
         cache/lru_cache_test.cc
-        db/blob_file_state_test.cc
+        db/blob_file_addition_test.cc
+        db/blob_file_garbage_test.cc
         db/column_family_test.cc
         db/compact_files_test.cc
         db/compaction/compaction_job_stats_test.cc
diff --git a/Makefile b/Makefile
index 6aef6c3c6..53d314b7c 100644
--- a/Makefile
+++ b/Makefile
@@ -597,7 +597,8 @@ TESTS = \
 	block_cache_tracer_test \
 	block_cache_trace_analyzer_test \
 	defer_test \
-	blob_file_state_test \
+	blob_file_addition_test \
+	blob_file_garbage_test \
 
 ifeq ($(USE_FOLLY_DISTRIBUTED_MUTEX),1)
 	TESTS += folly_synchronization_distributed_mutex_test
@@ -1719,7 +1720,10 @@ block_cache_trace_analyzer_test: tools/block_cache_analyzer/block_cache_trace_an
 defer_test: util/defer_test.o $(LIBOBJECTS) $(TESTHARNESS)
 	$(AM_LINK)
 
-blob_file_state_test: db/blob_file_state_test.o $(LIBOBJECTS) $(TESTHARNESS)
+blob_file_addition_test: db/blob_file_addition_test.o $(LIBOBJECTS) $(TESTHARNESS)
+	$(AM_LINK)
+
+blob_file_garbage_test: db/blob_file_garbage_test.o $(LIBOBJECTS) $(TESTHARNESS)
 	$(AM_LINK)
 
 #-------------------------------------------------
diff --git a/TARGETS b/TARGETS
index afd52bd88..91ab98fa6 100644
--- a/TARGETS
+++ b/TARGETS
@@ -116,7 +116,8 @@ cpp_library(
         "cache/lru_cache.cc",
         "cache/sharded_cache.cc",
         "db/arena_wrapped_db_iter.cc",
-        "db/blob_file_state.cc",
+        "db/blob_file_addition.cc",
+        "db/blob_file_garbage.cc",
         "db/builder.cc",
         "db/c.cc",
         "db/column_family.cc",
@@ -480,8 +481,15 @@ ROCKS_TESTS = [
         [],
     ],
     [
-        "blob_file_state_test",
-        "db/blob_file_state_test.cc",
+        "blob_file_addition_test",
+        "db/blob_file_addition_test.cc",
+        "serial",
+        [],
+        [],
+    ],
+    [
+        "blob_file_garbage_test",
+        "db/blob_file_garbage_test.cc",
         "serial",
         [],
         [],
diff --git a/db/blob_constants.h b/db/blob_constants.h
new file mode 100644
index 000000000..a5d09ac76
--- /dev/null
+++ b/db/blob_constants.h
@@ -0,0 +1,16 @@
+//  Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
+//  This source code is licensed under both the GPLv2 (found in the
+//  COPYING file in the root directory) and Apache 2.0 License
+//  (found in the LICENSE.Apache file in the root directory).
+
+#pragma once
+
+#include <cstdint>
+
+#include "rocksdb/rocksdb_namespace.h"
+
+namespace ROCKSDB_NAMESPACE {
+
+constexpr uint64_t kInvalidBlobFileNumber = 0;
+
+}  // namespace ROCKSDB_NAMESPACE
diff --git a/db/blob_file_state.cc b/db/blob_file_addition.cc
index 52a205c94..994b23b1e 100644
--- a/db/blob_file_state.cc
+++ b/db/blob_file_addition.cc
@@ -3,7 +3,10 @@
 //  COPYING file in the root directory) and Apache 2.0 License
 //  (found in the LICENSE.Apache file in the root directory).
 
-#include "db/blob_file_state.h"
+#include "db/blob_file_addition.h"
+
+#include <ostream>
+#include <sstream>
 
 #include "logging/event_logger.h"
 #include "rocksdb/slice.h"
@@ -11,9 +14,6 @@
 #include "test_util/sync_point.h"
 #include "util/coding.h"
 
-#include <ostream>
-#include <sstream>
-
 namespace ROCKSDB_NAMESPACE {
 
 namespace {
@@ -34,12 +34,10 @@ enum CustomFieldTags : uint32_t {
 
 }  // anonymous namespace
 
-void BlobFileState::EncodeTo(std::string* output) const {
+void BlobFileAddition::EncodeTo(std::string* output) const {
   PutVarint64(output, blob_file_number_);
   PutVarint64(output, total_blob_count_);
   PutVarint64(output, total_blob_bytes_);
-  PutVarint64(output, garbage_blob_count_);
-  PutVarint64(output, garbage_blob_bytes_);
   PutLengthPrefixedSlice(output, checksum_method_);
   PutLengthPrefixedSlice(output, checksum_value_);
 
@@ -48,13 +46,13 @@ void BlobFileState::EncodeTo(std::string* output) const {
   // fields will be ignored during decoding unless they're in the forward
   // incompatible range.
 
-  TEST_SYNC_POINT_CALLBACK("BlobFileState::EncodeTo::CustomFields", output);
+  TEST_SYNC_POINT_CALLBACK("BlobFileAddition::EncodeTo::CustomFields", output);
 
   PutVarint32(output, kEndMarker);
 }
 
-Status BlobFileState::DecodeFrom(Slice* input) {
-  constexpr char class_name[] = "BlobFileState";
+Status BlobFileAddition::DecodeFrom(Slice* input) {
+  constexpr char class_name[] = "BlobFileAddition";
 
   if (!GetVarint64(input, &blob_file_number_)) {
     return Status::Corruption(class_name, "Error decoding blob file number");
@@ -68,14 +66,6 @@ Status BlobFileState::DecodeFrom(Slice* input) {
     return Status::Corruption(class_name, "Error decoding total blob bytes");
   }
 
-  if (!GetVarint64(input, &garbage_blob_count_)) {
-    return Status::Corruption(class_name, "Error decoding garbage blob count");
-  }
-
-  if (!GetVarint64(input, &garbage_blob_bytes_)) {
-    return Status::Corruption(class_name, "Error decoding garbage blob bytes");
-  }
-
   Slice checksum_method;
   if (!GetLengthPrefixedSlice(input, &checksum_method)) {
     return Status::Corruption(class_name, "Error decoding checksum method");
@@ -113,7 +103,7 @@ Status BlobFileState::DecodeFrom(Slice* input) {
   return Status::OK();
 }
 
-std::string BlobFileState::DebugString() const {
+std::string BlobFileAddition::DebugString() const {
   std::ostringstream oss;
 
   oss << *this;
@@ -121,7 +111,7 @@ std::string BlobFileState::DebugString() const {
   return oss.str();
 }
 
-std::string BlobFileState::DebugJSON() const {
+std::string BlobFileAddition::DebugJSON() const {
   JSONWriter jw;
 
   jw << *this;
@@ -131,41 +121,36 @@ std::string BlobFileState::DebugJSON() const {
   return jw.Get();
 }
 
-bool operator==(const BlobFileState& lhs, const BlobFileState& rhs) {
+bool operator==(const BlobFileAddition& lhs, const BlobFileAddition& rhs) {
   return lhs.GetBlobFileNumber() == rhs.GetBlobFileNumber() &&
          lhs.GetTotalBlobCount() == rhs.GetTotalBlobCount() &&
          lhs.GetTotalBlobBytes() == rhs.GetTotalBlobBytes() &&
-         lhs.GetGarbageBlobCount() == rhs.GetGarbageBlobCount() &&
-         lhs.GetGarbageBlobBytes() == rhs.GetGarbageBlobBytes() &&
          lhs.GetChecksumMethod() == rhs.GetChecksumMethod() &&
          lhs.GetChecksumValue() == rhs.GetChecksumValue();
 }
 
-bool operator!=(const BlobFileState& lhs, const BlobFileState& rhs) {
+bool operator!=(const BlobFileAddition& lhs, const BlobFileAddition& rhs) {
   return !(lhs == rhs);
 }
 
 std::ostream& operator<<(std::ostream& os,
-                         const BlobFileState& blob_file_state) {
-  os << "blob_file_number: " << blob_file_state.GetBlobFileNumber()
-     << " total_blob_count: " << blob_file_state.GetTotalBlobCount()
-     << " total_blob_bytes: " << blob_file_state.GetTotalBlobBytes()
-     << " garbage_blob_count: " << blob_file_state.GetGarbageBlobCount()
-     << " garbage_blob_bytes: " << blob_file_state.GetGarbageBlobBytes()
-     << " checksum_method: " << blob_file_state.GetChecksumMethod()
-     << " checksum_value: " << blob_file_state.GetChecksumValue();
+                         const BlobFileAddition& blob_file_addition) {
+  os << "blob_file_number: " << blob_file_addition.GetBlobFileNumber()
+     << " total_blob_count: " << blob_file_addition.GetTotalBlobCount()
+     << " total_blob_bytes: " << blob_file_addition.GetTotalBlobBytes()
+     << " checksum_method: " << blob_file_addition.GetChecksumMethod()
+     << " checksum_value: " << blob_file_addition.GetChecksumValue();
 
   return os;
 }
 
-JSONWriter& operator<<(JSONWriter& jw, const BlobFileState& blob_file_state) {
-  jw << "BlobFileNumber" << blob_file_state.GetBlobFileNumber()
-     << "TotalBlobCount" << blob_file_state.GetTotalBlobCount()
-     << "TotalBlobBytes" << blob_file_state.GetTotalBlobBytes()
-     << "GarbageBlobCount" << blob_file_state.GetGarbageBlobCount()
-     << "GarbageBlobBytes" << blob_file_state.GetGarbageBlobBytes()
-     << "ChecksumMethod" << blob_file_state.GetChecksumMethod()
-     << "ChecksumValue" << blob_file_state.GetChecksumValue();
+JSONWriter& operator<<(JSONWriter& jw,
+                       const BlobFileAddition& blob_file_addition) {
+  jw << "BlobFileNumber" << blob_file_addition.GetBlobFileNumber()
+     << "TotalBlobCount" << blob_file_addition.GetTotalBlobCount()
+     << "TotalBlobBytes" << blob_file_addition.GetTotalBlobBytes()
+     << "ChecksumMethod" << blob_file_addition.GetChecksumMethod()
+     << "ChecksumValue" << blob_file_addition.GetChecksumValue();
 
   return jw;
 }
diff --git a/db/blob_file_addition.h b/db/blob_file_addition.h
new file mode 100644
index 000000000..8241038b2
--- /dev/null
+++ b/db/blob_file_addition.h
@@ -0,0 +1,65 @@
+//  Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
+//  This source code is licensed under both the GPLv2 (found in the
+//  COPYING file in the root directory) and Apache 2.0 License
+//  (found in the LICENSE.Apache file in the root directory).
+
+#pragma once
+
+#include <cassert>
+#include <cstdint>
+#include <iosfwd>
+#include <string>
+
+#include "db/blob_constants.h"
+#include "rocksdb/rocksdb_namespace.h"
+
+namespace ROCKSDB_NAMESPACE {
+
+class JSONWriter;
+class Slice;
+class Status;
+
+class BlobFileAddition {
+ public:
+  BlobFileAddition() = default;
+
+  BlobFileAddition(uint64_t blob_file_number, uint64_t total_blob_count,
+                   uint64_t total_blob_bytes, std::string checksum_method,
+                   std::string checksum_value)
+      : blob_file_number_(blob_file_number),
+        total_blob_count_(total_blob_count),
+        total_blob_bytes_(total_blob_bytes),
+        checksum_method_(std::move(checksum_method)),
+        checksum_value_(std::move(checksum_value)) {
+    assert(checksum_method_.empty() == checksum_value_.empty());
+  }
+
+  uint64_t GetBlobFileNumber() const { return blob_file_number_; }
+  uint64_t GetTotalBlobCount() const { return total_blob_count_; }
+  uint64_t GetTotalBlobBytes() const { return total_blob_bytes_; }
+  const std::string& GetChecksumMethod() const { return checksum_method_; }
+  const std::string& GetChecksumValue() const { return checksum_value_; }
+
+  void EncodeTo(std::string* output) const;
+  Status DecodeFrom(Slice* input);
+
+  std::string DebugString() const;
+  std::string DebugJSON() const;
+
+ private:
+  uint64_t blob_file_number_ = kInvalidBlobFileNumber;
+  uint64_t total_blob_count_ = 0;
+  uint64_t total_blob_bytes_ = 0;
+  std::string checksum_method_;
+  std::string checksum_value_;
+};
+
+bool operator==(const BlobFileAddition& lhs, const BlobFileAddition& rhs);
+bool operator!=(const BlobFileAddition& lhs, const BlobFileAddition& rhs);
+
+std::ostream& operator<<(std::ostream& os,
+                         const BlobFileAddition& blob_file_addition);
+JSONWriter& operator<<(JSONWriter& jw,
+                       const BlobFileAddition& blob_file_addition);
+
+}  // namespace ROCKSDB_NAMESPACE
diff --git a/db/blob_file_addition_test.cc b/db/blob_file_addition_test.cc
new file mode 100644
index 000000000..60eb756b2
--- /dev/null
+++ b/db/blob_file_addition_test.cc
@@ -0,0 +1,206 @@
+//  Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
+//  This source code is licensed under both the GPLv2 (found in the
+//  COPYING file in the root directory) and Apache 2.0 License
+//  (found in the LICENSE.Apache file in the root directory).
+
+#include "db/blob_file_addition.h"
+
+#include <cstdint>
+#include <cstring>
+#include <string>
+
+#include "test_util/sync_point.h"
+#include "test_util/testharness.h"
+#include "util/coding.h"
+
+namespace ROCKSDB_NAMESPACE {
+
+class BlobFileAdditionTest : public testing::Test {
+ public:
+  static void TestEncodeDecode(const BlobFileAddition& blob_file_addition) {
+    std::string encoded;
+    blob_file_addition.EncodeTo(&encoded);
+
+    BlobFileAddition decoded;
+    Slice input(encoded);
+    ASSERT_OK(decoded.DecodeFrom(&input));
+
+    ASSERT_EQ(blob_file_addition, decoded);
+  }
+};
+
+TEST_F(BlobFileAdditionTest, Empty) {
+  BlobFileAddition blob_file_addition;
+
+  ASSERT_EQ(blob_file_addition.GetBlobFileNumber(), kInvalidBlobFileNumber);
+  ASSERT_EQ(blob_file_addition.GetTotalBlobCount(), 0);
+  ASSERT_EQ(blob_file_addition.GetTotalBlobBytes(), 0);
+  ASSERT_TRUE(blob_file_addition.GetChecksumMethod().empty());
+  ASSERT_TRUE(blob_file_addition.GetChecksumValue().empty());
+
+  TestEncodeDecode(blob_file_addition);
+}
+
+TEST_F(BlobFileAdditionTest, NonEmpty) {
+  constexpr uint64_t blob_file_number = 123;
+  constexpr uint64_t total_blob_count = 2;
+  constexpr uint64_t total_blob_bytes = 123456;
+  const std::string checksum_method("SHA1");
+  const std::string checksum_value("bdb7f34a59dfa1592ce7f52e99f98c570c525cbd");
+
+  BlobFileAddition blob_file_addition(blob_file_number, total_blob_count,
+                                      total_blob_bytes, checksum_method,
+                                      checksum_value);
+
+  ASSERT_EQ(blob_file_addition.GetBlobFileNumber(), blob_file_number);
+  ASSERT_EQ(blob_file_addition.GetTotalBlobCount(), total_blob_count);
+  ASSERT_EQ(blob_file_addition.GetTotalBlobBytes(), total_blob_bytes);
+  ASSERT_EQ(blob_file_addition.GetChecksumMethod(), checksum_method);
+  ASSERT_EQ(blob_file_addition.GetChecksumValue(), checksum_value);
+
+  TestEncodeDecode(blob_file_addition);
+}
+
+TEST_F(BlobFileAdditionTest, DecodeErrors) {
+  std::string str;
+  Slice slice(str);
+
+  BlobFileAddition blob_file_addition;
+
+  {
+    const Status s = blob_file_addition.DecodeFrom(&slice);
+    ASSERT_TRUE(s.IsCorruption());
+    ASSERT_TRUE(std::strstr(s.getState(), "blob file number"));
+  }
+
+  constexpr uint64_t blob_file_number = 123;
+  PutVarint64(&str, blob_file_number);
+  slice = str;
+
+  {
+    const Status s = blob_file_addition.DecodeFrom(&slice);
+    ASSERT_TRUE(s.IsCorruption());
+    ASSERT_TRUE(std::strstr(s.getState(), "total blob count"));
+  }
+
+  constexpr uint64_t total_blob_count = 4567;
+  PutVarint64(&str, total_blob_count);
+  slice = str;
+
+  {
+    const Status s = blob_file_addition.DecodeFrom(&slice);
+    ASSERT_TRUE(s.IsCorruption());
+    ASSERT_TRUE(std::strstr(s.getState(), "total blob bytes"));
+  }
+
+  constexpr uint64_t total_blob_bytes = 12345678;
+  PutVarint64(&str, total_blob_bytes);
+  slice = str;
+
+  {
+    const Status s = blob_file_addition.DecodeFrom(&slice);
+    ASSERT_TRUE(s.IsCorruption());
+    ASSERT_TRUE(std::strstr(s.getState(), "checksum method"));
+  }
+
+  constexpr char checksum_method[] = "SHA1";
+  PutLengthPrefixedSlice(&str, checksum_method);
+  slice = str;
+
+  {
+    const Status s = blob_file_addition.DecodeFrom(&slice);
+    ASSERT_TRUE(s.IsCorruption());
+    ASSERT_TRUE(std::strstr(s.getState(), "checksum value"));
+  }
+
+  constexpr char checksum_value[] = "bdb7f34a59dfa1592ce7f52e99f98c570c525cbd";
+  PutLengthPrefixedSlice(&str, checksum_value);
+  slice = str;
+
+  {
+    const Status s = blob_file_addition.DecodeFrom(&slice);
+    ASSERT_TRUE(s.IsCorruption());
+    ASSERT_TRUE(std::strstr(s.getState(), "custom field tag"));
+  }
+
+  constexpr uint32_t custom_tag = 2;
+  PutVarint32(&str, custom_tag);
+  slice = str;
+
+  {
+    const Status s = blob_file_addition.DecodeFrom(&slice);
+    ASSERT_TRUE(s.IsCorruption());
+    ASSERT_TRUE(std::strstr(s.getState(), "custom field value"));
+  }
+}
+
+TEST_F(BlobFileAdditionTest, ForwardCompatibleCustomField) {
+  SyncPoint::GetInstance()->SetCallBack(
+      "BlobFileAddition::EncodeTo::CustomFields", [&](void* arg) {
+        std::string* output = static_cast<std::string*>(arg);
+
+        constexpr uint32_t forward_compatible_tag = 2;
+        PutVarint32(output, forward_compatible_tag);
+
+        PutLengthPrefixedSlice(output, "deadbeef");
+      });
+  SyncPoint::GetInstance()->EnableProcessing();
+
+  constexpr uint64_t blob_file_number = 678;
+  constexpr uint64_t total_blob_count = 9999;
+  constexpr uint64_t total_blob_bytes = 100000000;
+  const std::string checksum_method("CRC32");
+  const std::string checksum_value("3d87ff57");
+
+  BlobFileAddition blob_file_addition(blob_file_number, total_blob_count,
+                                      total_blob_bytes, checksum_method,
+                                      checksum_value);
+
+  TestEncodeDecode(blob_file_addition);
+
+  SyncPoint::GetInstance()->DisableProcessing();
+  SyncPoint::GetInstance()->ClearAllCallBacks();
+}
+
+TEST_F(BlobFileAdditionTest, ForwardIncompatibleCustomField) {
+  SyncPoint::GetInstance()->SetCallBack(
+      "BlobFileAddition::EncodeTo::CustomFields", [&](void* arg) {
+        std::string* output = static_cast<std::string*>(arg);
+
+        constexpr uint32_t forward_incompatible_tag = (1 << 6) + 1;
+        PutVarint32(output, forward_incompatible_tag);
+
+        PutLengthPrefixedSlice(output, "foobar");
+      });
+  SyncPoint::GetInstance()->EnableProcessing();
+
+  constexpr uint64_t blob_file_number = 456;
+  constexpr uint64_t total_blob_count = 100;
+  constexpr uint64_t total_blob_bytes = 2000000;
+  const std::string checksum_method("CRC32B");
+  const std::string checksum_value("6dbdf23a");
+
+  BlobFileAddition blob_file_addition(blob_file_number, total_blob_count,
+                                      total_blob_bytes, checksum_method,
+                                      checksum_value);
+
+  std::string encoded;
+  blob_file_addition.EncodeTo(&encoded);
+
+  BlobFileAddition decoded_blob_file_addition;
+  Slice input(encoded);
+  const Status s = decoded_blob_file_addition.DecodeFrom(&input);
+
+  ASSERT_TRUE(s.IsCorruption());
+  ASSERT_TRUE(std::strstr(s.getState(), "Forward incompatible"));
+
+  SyncPoint::GetInstance()->DisableProcessing();
+  SyncPoint::GetInstance()->ClearAllCallBacks();
+}
+
+}  // namespace ROCKSDB_NAMESPACE
+
+int main(int argc, char** argv) {
+  ::testing::InitGoogleTest(&argc, argv);
+  return RUN_ALL_TESTS();
+}
diff --git a/db/blob_file_garbage.cc b/db/blob_file_garbage.cc
new file mode 100644
index 000000000..2d399bf44
--- /dev/null
+++ b/db/blob_file_garbage.cc
@@ -0,0 +1,138 @@
+//  Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
+//  This source code is licensed under both the GPLv2 (found in the
+//  COPYING file in the root directory) and Apache 2.0 License
+//  (found in the LICENSE.Apache file in the root directory).
+
+#include "db/blob_file_garbage.h"
+
+#include <ostream>
+#include <sstream>
+
+#include "logging/event_logger.h"
+#include "rocksdb/slice.h"
+#include "rocksdb/status.h"
+#include "test_util/sync_point.h"
+#include "util/coding.h"
+
+namespace ROCKSDB_NAMESPACE {
+
+namespace {
+
+// Tags for custom fields. Note that these get persisted in the manifest,
+// so existing tags should not be modified.
+enum CustomFieldTags : uint32_t {
+  kEndMarker,
+
+  // Add forward compatible fields here
+
+  /////////////////////////////////////////////////////////////////////
+
+  kForwardIncompatibleMask = 1 << 6,
+
+  // Add forward incompatible fields here
+};
+
+}  // anonymous namespace
+
+void BlobFileGarbage::EncodeTo(std::string* output) const {
+  PutVarint64(output, blob_file_number_);
+  PutVarint64(output, garbage_blob_count_);
+  PutVarint64(output, garbage_blob_bytes_);
+
+  // Encode any custom fields here. The format to use is a Varint32 tag (see
+  // CustomFieldTags above) followed by a length prefixed slice. Unknown custom
+  // fields will be ignored during decoding unless they're in the forward
+  // incompatible range.
+
+  TEST_SYNC_POINT_CALLBACK("BlobFileGarbage::EncodeTo::CustomFields", output);
+
+  PutVarint32(output, kEndMarker);
+}
+
+Status BlobFileGarbage::DecodeFrom(Slice* input) {
+  constexpr char class_name[] = "BlobFileGarbage";
+
+  if (!GetVarint64(input, &blob_file_number_)) {
+    return Status::Corruption(class_name, "Error decoding blob file number");
+  }
+
+  if (!GetVarint64(input, &garbage_blob_count_)) {
+    return Status::Corruption(class_name, "Error decoding garbage blob count");
+  }
+
+  if (!GetVarint64(input, &garbage_blob_bytes_)) {
+    return Status::Corruption(class_name, "Error decoding garbage blob bytes");
+  }
+
+  while (true) {
+    uint32_t custom_field_tag = 0;
+    if (!GetVarint32(input, &custom_field_tag)) {
+      return Status::Corruption(class_name, "Error decoding custom field tag");
+    }
+
+    if (custom_field_tag == kEndMarker) {
+      break;
+    }
+
+    if (custom_field_tag & kForwardIncompatibleMask) {
+      return Status::Corruption(
+          class_name, "Forward incompatible custom field encountered");
+    }
+
+    Slice custom_field_value;
+    if (!GetLengthPrefixedSlice(input, &custom_field_value)) {
+      return Status::Corruption(class_name,
+                                "Error decoding custom field value");
+    }
+  }
+
+  return Status::OK();
+}
+
+std::string BlobFileGarbage::DebugString() const {
+  std::ostringstream oss;
+
+  oss << *this;
+
+  return oss.str();
+}
+
+std::string BlobFileGarbage::DebugJSON() const {
+  JSONWriter jw;
+
+  jw << *this;
+
+  jw.EndObject();
+
+  return jw.Get();
+}
+
+bool operator==(const BlobFileGarbage& lhs, const BlobFileGarbage& rhs) {
+  return lhs.GetBlobFileNumber() == rhs.GetBlobFileNumber() &&
+         lhs.GetGarbageBlobCount() == rhs.GetGarbageBlobCount() &&
+         lhs.GetGarbageBlobBytes() == rhs.GetGarbageBlobBytes();
+}
+
+bool operator!=(const BlobFileGarbage& lhs, const BlobFileGarbage& rhs) {
+  return !(lhs == rhs);
+}
+
+std::ostream& operator<<(std::ostream& os,
+                         const BlobFileGarbage& blob_file_garbage) {
+  os << "blob_file_number: " << blob_file_garbage.GetBlobFileNumber()
+     << " garbage_blob_count: " << blob_file_garbage.GetGarbageBlobCount()
+     << " garbage_blob_bytes: " << blob_file_garbage.GetGarbageBlobBytes();
+
+  return os;
+}
+
+JSONWriter& operator<<(JSONWriter& jw,
+                       const BlobFileGarbage& blob_file_garbage) {
+  jw << "BlobFileNumber" << blob_file_garbage.GetBlobFileNumber()
+     << "GarbageBlobCount" << blob_file_garbage.GetGarbageBlobCount()
+     << "GarbageBlobBytes" << blob_file_garbage.GetGarbageBlobBytes();
+
+  return jw;
+}
+
+}  // namespace ROCKSDB_NAMESPACE
diff --git a/db/blob_file_garbage.h b/db/blob_file_garbage.h
new file mode 100644
index 000000000..ae6c23231
--- /dev/null
+++ b/db/blob_file_garbage.h
@@ -0,0 +1,55 @@
+//  Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
+//  This source code is licensed under both the GPLv2 (found in the
+//  COPYING file in the root directory) and Apache 2.0 License
+//  (found in the LICENSE.Apache file in the root directory).
+
+#pragma once
+
+#include <cstdint>
+#include <iosfwd>
+#include <string>
+
+#include "db/blob_constants.h"
+#include "rocksdb/rocksdb_namespace.h"
+
+namespace ROCKSDB_NAMESPACE {
+
+class JSONWriter;
+class Slice;
+class Status;
+
+class BlobFileGarbage {
+ public:
+  BlobFileGarbage() = default;
+
+  BlobFileGarbage(uint64_t blob_file_number, uint64_t garbage_blob_count,
+                  uint64_t garbage_blob_bytes)
+      : blob_file_number_(blob_file_number),
+        garbage_blob_count_(garbage_blob_count),
+        garbage_blob_bytes_(garbage_blob_bytes) {}
+
+  uint64_t GetBlobFileNumber() const { return blob_file_number_; }
+  uint64_t GetGarbageBlobCount() const { return garbage_blob_count_; }
+  uint64_t GetGarbageBlobBytes() const { return garbage_blob_bytes_; }
+
+  void EncodeTo(std::string* output) const;
+  Status DecodeFrom(Slice* input);
+
+  std::string DebugString() const;
+  std::string DebugJSON() const;
+
+ private:
+  uint64_t blob_file_number_ = kInvalidBlobFileNumber;
+  uint64_t garbage_blob_count_ = 0;
+  uint64_t garbage_blob_bytes_ = 0;
+};
+
+bool operator==(const BlobFileGarbage& lhs, const BlobFileGarbage& rhs);
+bool operator!=(const BlobFileGarbage& lhs, const BlobFileGarbage& rhs);
+
+std::ostream& operator<<(std::ostream& os,
+                         const BlobFileGarbage& blob_file_garbage);
+JSONWriter& operator<<(JSONWriter& jw,
+                       const BlobFileGarbage& blob_file_garbage);
+
+}  // namespace ROCKSDB_NAMESPACE
diff --git a/db/blob_file_garbage_test.cc b/db/blob_file_garbage_test.cc
new file mode 100644
index 000000000..c5f4c022d
--- /dev/null
+++ b/db/blob_file_garbage_test.cc
@@ -0,0 +1,173 @@
+//  Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
+//  This source code is licensed under both the GPLv2 (found in the
+//  COPYING file in the root directory) and Apache 2.0 License
+//  (found in the LICENSE.Apache file in the root directory).
+
+#include "db/blob_file_garbage.h"
+
+#include <cstdint>
+#include <cstring>
+#include <string>
+
+#include "test_util/sync_point.h"
+#include "test_util/testharness.h"
+#include "util/coding.h"
+
+namespace ROCKSDB_NAMESPACE {
+
+class BlobFileGarbageTest : public testing::Test {
+ public:
+  static void TestEncodeDecode(const BlobFileGarbage& blob_file_garbage) {
+    std::string encoded;
+    blob_file_garbage.EncodeTo(&encoded);
+
+    BlobFileGarbage decoded;
+    Slice input(encoded);
+    ASSERT_OK(decoded.DecodeFrom(&input));
+
+    ASSERT_EQ(blob_file_garbage, decoded);
+  }
+};
+
+TEST_F(BlobFileGarbageTest, Empty) {
+  BlobFileGarbage blob_file_garbage;
+
+  ASSERT_EQ(blob_file_garbage.GetBlobFileNumber(), kInvalidBlobFileNumber);
+  ASSERT_EQ(blob_file_garbage.GetGarbageBlobCount(), 0);
+  ASSERT_EQ(blob_file_garbage.GetGarbageBlobBytes(), 0);
+
+  TestEncodeDecode(blob_file_garbage);
+}
+
+TEST_F(BlobFileGarbageTest, NonEmpty) {
+  constexpr uint64_t blob_file_number = 123;
+  constexpr uint64_t garbage_blob_count = 1;
+  constexpr uint64_t garbage_blob_bytes = 9876;
+
+  BlobFileGarbage blob_file_garbage(blob_file_number, garbage_blob_count,
+                                    garbage_blob_bytes);
+
+  ASSERT_EQ(blob_file_garbage.GetBlobFileNumber(), blob_file_number);
+  ASSERT_EQ(blob_file_garbage.GetGarbageBlobCount(), garbage_blob_count);
+  ASSERT_EQ(blob_file_garbage.GetGarbageBlobBytes(), garbage_blob_bytes);
+
+  TestEncodeDecode(blob_file_garbage);
+}
+
+TEST_F(BlobFileGarbageTest, DecodeErrors) {
+  std::string str;
+  Slice slice(str);
+
+  BlobFileGarbage blob_file_garbage;
+
+  {
+    const Status s = blob_file_garbage.DecodeFrom(&slice);
+    ASSERT_TRUE(s.IsCorruption());
+    ASSERT_TRUE(std::strstr(s.getState(), "blob file number"));
+  }
+
+  constexpr uint64_t blob_file_number = 123;
+  PutVarint64(&str, blob_file_number);
+  slice = str;
+
+  {
+    const Status s = blob_file_garbage.DecodeFrom(&slice);
+    ASSERT_TRUE(s.IsCorruption());
+    ASSERT_TRUE(std::strstr(s.getState(), "garbage blob count"));
+  }
+
+  constexpr uint64_t garbage_blob_count = 4567;
+  PutVarint64(&str, garbage_blob_count);
+  slice = str;
+
+  {
+    const Status s = blob_file_garbage.DecodeFrom(&slice);
+    ASSERT_TRUE(s.IsCorruption());
+    ASSERT_TRUE(std::strstr(s.getState(), "garbage blob bytes"));
+  }
+
+  constexpr uint64_t garbage_blob_bytes = 12345678;
+  PutVarint64(&str, garbage_blob_bytes);
+  slice = str;
+
+  {
+    const Status s = blob_file_garbage.DecodeFrom(&slice);
+    ASSERT_TRUE(s.IsCorruption());
+    ASSERT_TRUE(std::strstr(s.getState(), "custom field tag"));
+  }
+
+  constexpr uint32_t custom_tag = 2;
+  PutVarint32(&str, custom_tag);
+  slice = str;
+
+  {
+    const Status s = blob_file_garbage.DecodeFrom(&slice);
+    ASSERT_TRUE(s.IsCorruption());
+    ASSERT_TRUE(std::strstr(s.getState(), "custom field value"));
+  }
+}
+
+TEST_F(BlobFileGarbageTest, ForwardCompatibleCustomField) {
+  SyncPoint::GetInstance()->SetCallBack(
+      "BlobFileGarbage::EncodeTo::CustomFields", [&](void* arg) {
+        std::string* output = static_cast<std::string*>(arg);
+
+        constexpr uint32_t forward_compatible_tag = 2;
+        PutVarint32(output, forward_compatible_tag);
+
+        PutLengthPrefixedSlice(output, "deadbeef");
+      });
+  SyncPoint::GetInstance()->EnableProcessing();
+
+  constexpr uint64_t blob_file_number = 678;
+  constexpr uint64_t garbage_blob_count = 9999;
+  constexpr uint64_t garbage_blob_bytes = 100000000;
+
+  BlobFileGarbage blob_file_garbage(blob_file_number, garbage_blob_count,
+                                    garbage_blob_bytes);
+
+  TestEncodeDecode(blob_file_garbage);
+
+  SyncPoint::GetInstance()->DisableProcessing();
+  SyncPoint::GetInstance()->ClearAllCallBacks();
+}
+
+TEST_F(BlobFileGarbageTest, ForwardIncompatibleCustomField) {
+  SyncPoint::GetInstance()->SetCallBack(
+      "BlobFileGarbage::EncodeTo::CustomFields", [&](void* arg) {
+        std::string* output = static_cast<std::string*>(arg);
+
+        constexpr uint32_t forward_incompatible_tag = (1 << 6) + 1;
+        PutVarint32(output, forward_incompatible_tag);
+
+        PutLengthPrefixedSlice(output, "foobar");
+      });
+  SyncPoint::GetInstance()->EnableProcessing();
+
+  constexpr uint64_t blob_file_number = 456;
+  constexpr uint64_t garbage_blob_count = 100;
+  constexpr uint64_t garbage_blob_bytes = 2000000;
+
+  BlobFileGarbage blob_file_garbage(blob_file_number, garbage_blob_count,
+                                    garbage_blob_bytes);
+
+  std::string encoded;
+  blob_file_garbage.EncodeTo(&encoded);
+
+  BlobFileGarbage decoded_blob_file_addition;
+  Slice input(encoded);
+  const Status s = decoded_blob_file_addition.DecodeFrom(&input);
+
+  ASSERT_TRUE(s.IsCorruption());
+  ASSERT_TRUE(std::strstr(s.getState(), "Forward incompatible"));
+
+  SyncPoint::GetInstance()->DisableProcessing();
+  SyncPoint::GetInstance()->ClearAllCallBacks();
+}
+
+}  // namespace ROCKSDB_NAMESPACE
+
+int main(int argc, char** argv) {
+  ::testing::InitGoogleTest(&argc, argv);
+  return RUN_ALL_TESTS();
+}
diff --git a/db/blob_file_state.h b/db/blob_file_state.h
deleted file mode 100644
index 0591d52d2..000000000
--- a/db/blob_file_state.h
+++ /dev/null
@@ -1,102 +0,0 @@
-//  Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
-//  This source code is licensed under both the GPLv2 (found in the
-//  COPYING file in the root directory) and Apache 2.0 License
-//  (found in the LICENSE.Apache file in the root directory).
-
-#pragma once
-
-#include "rocksdb/rocksdb_namespace.h"
-
-#include <cassert>
-#include <cstdint>
-#include <iosfwd>
-#include <string>
-
-namespace ROCKSDB_NAMESPACE {
-
-constexpr uint64_t kInvalidBlobFileNumber = 0;
-
-class JSONWriter;
-class Slice;
-class Status;
-
-class BlobFileState {
- public:
-  BlobFileState() = default;
-
-  BlobFileState(uint64_t blob_file_number, uint64_t total_blob_count,
-                uint64_t total_blob_bytes, std::string checksum_method,
-                std::string checksum_value)
-      : blob_file_number_(blob_file_number),
-        total_blob_count_(total_blob_count),
-        total_blob_bytes_(total_blob_bytes),
-        checksum_method_(std::move(checksum_method)),
-        checksum_value_(std::move(checksum_value)) {
-    assert(checksum_method_.empty() == checksum_value_.empty());
-  }
-
-  BlobFileState(uint64_t blob_file_number, uint64_t total_blob_count,
-                uint64_t total_blob_bytes, uint64_t garbage_blob_count,
-                uint64_t garbage_blob_bytes, std::string checksum_method,
-                std::string checksum_value)
-      : blob_file_number_(blob_file_number),
-        total_blob_count_(total_blob_count),
-        total_blob_bytes_(total_blob_bytes),
-        garbage_blob_count_(garbage_blob_count),
-        garbage_blob_bytes_(garbage_blob_bytes),
-        checksum_method_(std::move(checksum_method)),
-        checksum_value_(std::move(checksum_value)) {
-    assert(checksum_method_.empty() == checksum_value_.empty());
-    assert(garbage_blob_count_ <= total_blob_count_);
-    assert(garbage_blob_bytes_ <= total_blob_bytes_);
-  }
-
-  uint64_t GetBlobFileNumber() const { return blob_file_number_; }
-
-  uint64_t GetTotalBlobCount() const { return total_blob_count_; }
-  uint64_t GetTotalBlobBytes() const { return total_blob_bytes_; }
-
-  void AddGarbageBlob(uint64_t size) {
-    assert(garbage_blob_count_ < total_blob_count_);
-    assert(garbage_blob_bytes_ + size <= total_blob_bytes_);
-
-    ++garbage_blob_count_;
-    garbage_blob_bytes_ += size;
-  }
-
-  uint64_t GetGarbageBlobCount() const { return garbage_blob_count_; }
-  uint64_t GetGarbageBlobBytes() const { return garbage_blob_bytes_; }
-
-  bool IsObsolete() const {
-    assert(garbage_blob_count_ <= total_blob_count_);
-
-    return !(garbage_blob_count_ < total_blob_count_);
-  }
-
-  const std::string& GetChecksumMethod() const { return checksum_method_; }
-  const std::string& GetChecksumValue() const { return checksum_value_; }
-
-  void EncodeTo(std::string* output) const;
-  Status DecodeFrom(Slice* input);
-
-  std::string DebugString() const;
-  std::string DebugJSON() const;
-
- private:
-  uint64_t blob_file_number_ = kInvalidBlobFileNumber;
-  uint64_t total_blob_count_ = 0;
-  uint64_t total_blob_bytes_ = 0;
-  uint64_t garbage_blob_count_ = 0;
-  uint64_t garbage_blob_bytes_ = 0;
-  std::string checksum_method_;
-  std::string checksum_value_;
-};
-
-bool operator==(const BlobFileState& lhs, const BlobFileState& rhs);
-bool operator!=(const BlobFileState& lhs, const BlobFileState& rhs);
-
-std::ostream& operator<<(std::ostream& os,
-                         const BlobFileState& blob_file_state);
-JSONWriter& operator<<(JSONWriter& jw, const BlobFileState& blob_file_state);
-
-}  // namespace ROCKSDB_NAMESPACE
diff --git a/db/blob_file_state_test.cc b/db/blob_file_state_test.cc
deleted file mode 100644
index b6cd3f6f9..000000000
--- a/db/blob_file_state_test.cc
+++ /dev/null
@@ -1,284 +0,0 @@
-//  Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
-//  This source code is licensed under both the GPLv2 (found in the
-//  COPYING file in the root directory) and Apache 2.0 License
-//  (found in the LICENSE.Apache file in the root directory).
-
-#include "db/blob_file_state.h"
-#include "test_util/sync_point.h"
-#include "test_util/testharness.h"
-#include "util/coding.h"
-
-#include <cstdint>
-#include <cstring>
-#include <string>
-
-namespace ROCKSDB_NAMESPACE {
-
-class BlobFileStateTest : public testing::Test {
- public:
-  static void TestEncodeDecode(const BlobFileState& blob_file_state) {
-    std::string encoded;
-    blob_file_state.EncodeTo(&encoded);
-
-    BlobFileState decoded;
-    Slice input(encoded);
-    ASSERT_OK(decoded.DecodeFrom(&input));
-
-    ASSERT_EQ(blob_file_state, decoded);
-  }
-};
-
-TEST_F(BlobFileStateTest, Empty) {
-  BlobFileState blob_file_state;
-
-  ASSERT_EQ(blob_file_state.GetBlobFileNumber(), kInvalidBlobFileNumber);
-  ASSERT_EQ(blob_file_state.GetTotalBlobCount(), 0);
-  ASSERT_EQ(blob_file_state.GetTotalBlobBytes(), 0);
-  ASSERT_EQ(blob_file_state.GetGarbageBlobCount(), 0);
-  ASSERT_EQ(blob_file_state.GetGarbageBlobBytes(), 0);
-  ASSERT_TRUE(blob_file_state.IsObsolete());
-  ASSERT_TRUE(blob_file_state.GetChecksumMethod().empty());
-  ASSERT_TRUE(blob_file_state.GetChecksumValue().empty());
-
-  TestEncodeDecode(blob_file_state);
-}
-
-TEST_F(BlobFileStateTest, NonEmpty) {
-  constexpr uint64_t blob_file_number = 123;
-  constexpr uint64_t total_blob_count = 2;
-  constexpr uint64_t total_blob_bytes = 123456;
-  constexpr uint64_t garbage_blob_count = 1;
-  constexpr uint64_t garbage_blob_bytes = 9876;
-  const std::string checksum_method("SHA1");
-  const std::string checksum_value("bdb7f34a59dfa1592ce7f52e99f98c570c525cbd");
-
-  BlobFileState blob_file_state(
-      blob_file_number, total_blob_count, total_blob_bytes, garbage_blob_count,
-      garbage_blob_bytes, checksum_method, checksum_value);
-
-  ASSERT_EQ(blob_file_state.GetBlobFileNumber(), blob_file_number);
-  ASSERT_EQ(blob_file_state.GetTotalBlobCount(), total_blob_count);
-  ASSERT_EQ(blob_file_state.GetTotalBlobBytes(), total_blob_bytes);
-  ASSERT_EQ(blob_file_state.GetGarbageBlobCount(), garbage_blob_count);
-  ASSERT_EQ(blob_file_state.GetGarbageBlobBytes(), garbage_blob_bytes);
-  ASSERT_FALSE(blob_file_state.IsObsolete());
-  ASSERT_EQ(blob_file_state.GetChecksumMethod(), checksum_method);
-  ASSERT_EQ(blob_file_state.GetChecksumValue(), checksum_value);
-
-  TestEncodeDecode(blob_file_state);
-}
-
-TEST_F(BlobFileStateTest, AddGarbageBlob) {
-  constexpr uint64_t blob_file_number = 123;
-  constexpr uint64_t total_blob_count = 2;
-  constexpr uint64_t total_blob_bytes = 123456;
-  const std::string checksum_method("MD5");
-  const std::string checksum_value("d8f72233c67a68c5ec2bd51c6be7556e");
-
-  BlobFileState blob_file_state(blob_file_number, total_blob_count,
-                                total_blob_bytes, checksum_method,
-                                checksum_value);
-
-  ASSERT_EQ(blob_file_state.GetBlobFileNumber(), blob_file_number);
-  ASSERT_EQ(blob_file_state.GetTotalBlobCount(), total_blob_count);
-  ASSERT_EQ(blob_file_state.GetTotalBlobBytes(), total_blob_bytes);
-  ASSERT_EQ(blob_file_state.GetGarbageBlobCount(), 0);
-  ASSERT_EQ(blob_file_state.GetGarbageBlobBytes(), 0);
-  ASSERT_FALSE(blob_file_state.IsObsolete());
-  ASSERT_EQ(blob_file_state.GetChecksumMethod(), checksum_method);
-  ASSERT_EQ(blob_file_state.GetChecksumValue(), checksum_value);
-
-  TestEncodeDecode(blob_file_state);
-
-  blob_file_state.AddGarbageBlob(123000);
-
-  ASSERT_EQ(blob_file_state.GetBlobFileNumber(), blob_file_number);
-  ASSERT_EQ(blob_file_state.GetTotalBlobCount(), total_blob_count);
-  ASSERT_EQ(blob_file_state.GetTotalBlobBytes(), total_blob_bytes);
-  ASSERT_EQ(blob_file_state.GetGarbageBlobCount(), 1);
-  ASSERT_EQ(blob_file_state.GetGarbageBlobBytes(), 123000);
-  ASSERT_FALSE(blob_file_state.IsObsolete());
-  ASSERT_EQ(blob_file_state.GetChecksumMethod(), checksum_method);
-  ASSERT_EQ(blob_file_state.GetChecksumValue(), checksum_value);
-
-  TestEncodeDecode(blob_file_state);
-
-  blob_file_state.AddGarbageBlob(456);
-
-  ASSERT_EQ(blob_file_state.GetBlobFileNumber(), blob_file_number);
-  ASSERT_EQ(blob_file_state.GetTotalBlobCount(), total_blob_count);
-  ASSERT_EQ(blob_file_state.GetTotalBlobBytes(), total_blob_bytes);
-  ASSERT_EQ(blob_file_state.GetGarbageBlobCount(), total_blob_count);
-  ASSERT_EQ(blob_file_state.GetGarbageBlobBytes(), total_blob_bytes);
-  ASSERT_TRUE(blob_file_state.IsObsolete());
-  ASSERT_EQ(blob_file_state.GetChecksumMethod(), checksum_method);
-  ASSERT_EQ(blob_file_state.GetChecksumValue(), checksum_value);
-
-  TestEncodeDecode(blob_file_state);
-}
-
-TEST_F(BlobFileStateTest, DecodeErrors) {
-  std::string str;
-  Slice slice(str);
-
-  BlobFileState blob_file_state;
-
-  {
-    const Status s = blob_file_state.DecodeFrom(&slice);
-    ASSERT_TRUE(s.IsCorruption());
-    ASSERT_TRUE(std::strstr(s.getState(), "blob file number"));
-  }
-
-  constexpr uint64_t blob_file_number = 123;
-  PutVarint64(&str, blob_file_number);
-  slice = str;
-
-  {
-    const Status s = blob_file_state.DecodeFrom(&slice);
-    ASSERT_TRUE(s.IsCorruption());
-    ASSERT_TRUE(std::strstr(s.getState(), "total blob count"));
-  }
-
-  constexpr uint64_t total_blob_count = 4567;
-  PutVarint64(&str, total_blob_count);
-  slice = str;
-
-  {
-    const Status s = blob_file_state.DecodeFrom(&slice);
-    ASSERT_TRUE(s.IsCorruption());
-    ASSERT_TRUE(std::strstr(s.getState(), "total blob bytes"));
-  }
-
-  constexpr uint64_t total_blob_bytes = 12345678;
-  PutVarint64(&str, total_blob_bytes);
-  slice = str;
-
-  {
-    const Status s = blob_file_state.DecodeFrom(&slice);
-    ASSERT_TRUE(s.IsCorruption());
-    ASSERT_TRUE(std::strstr(s.getState(), "garbage blob count"));
-  }
-
-  constexpr uint64_t garbage_blob_count = 1234;
-  PutVarint64(&str, garbage_blob_count);
-  slice = str;
-
-  {
-    const Status s = blob_file_state.DecodeFrom(&slice);
-    ASSERT_TRUE(s.IsCorruption());
-    ASSERT_TRUE(std::strstr(s.getState(), "garbage blob bytes"));
-  }
-
-  constexpr uint64_t garbage_blob_bytes = 5678;
-  PutVarint64(&str, garbage_blob_bytes);
-  slice = str;
-
-  {
-    const Status s = blob_file_state.DecodeFrom(&slice);
-    ASSERT_TRUE(s.IsCorruption());
-    ASSERT_TRUE(std::strstr(s.getState(), "checksum method"));
-  }
-
-  constexpr char checksum_method[] = "SHA1";
-  PutLengthPrefixedSlice(&str, checksum_method);
-  slice = str;
-
-  {
-    const Status s = blob_file_state.DecodeFrom(&slice);
-    ASSERT_TRUE(s.IsCorruption());
-    ASSERT_TRUE(std::strstr(s.getState(), "checksum value"));
-  }
-
-  constexpr char checksum_value[] = "bdb7f34a59dfa1592ce7f52e99f98c570c525cbd";
-  PutLengthPrefixedSlice(&str, checksum_value);
-  slice = str;
-
-  {
-    const Status s = blob_file_state.DecodeFrom(&slice);
-    ASSERT_TRUE(s.IsCorruption());
-    ASSERT_TRUE(std::strstr(s.getState(), "custom field tag"));
-  }
-
-  constexpr uint32_t custom_tag = 2;
-  PutVarint32(&str, custom_tag);
-  slice = str;
-
-  {
-    const Status s = blob_file_state.DecodeFrom(&slice);
-    ASSERT_TRUE(s.IsCorruption());
-    ASSERT_TRUE(std::strstr(s.getState(), "custom field value"));
-  }
-}
-
-TEST_F(BlobFileStateTest, ForwardCompatibleCustomField) {
-  SyncPoint::GetInstance()->SetCallBack(
-      "BlobFileState::EncodeTo::CustomFields", [&](void* arg) {
-        std::string* output = static_cast<std::string*>(arg);
-
-        constexpr uint32_t forward_compatible_tag = 2;
-        PutVarint32(output, forward_compatible_tag);
-
-        PutLengthPrefixedSlice(output, "deadbeef");
-      });
-  SyncPoint::GetInstance()->EnableProcessing();
-
-  constexpr uint64_t blob_file_number = 678;
-  constexpr uint64_t total_blob_count = 9999;
-  constexpr uint64_t total_blob_bytes = 100000000;
-  constexpr uint64_t garbage_blob_count = 3333;
-  constexpr uint64_t garbage_blob_bytes = 2500000;
-  const std::string checksum_method("CRC32");
-  const std::string checksum_value("3d87ff57");
-
-  BlobFileState blob_file_state(
-      blob_file_number, total_blob_count, total_blob_bytes, garbage_blob_count,
-      garbage_blob_bytes, checksum_method, checksum_value);
-
-  TestEncodeDecode(blob_file_state);
-
-  SyncPoint::GetInstance()->DisableProcessing();
-  SyncPoint::GetInstance()->ClearAllCallBacks();
-}
-
-TEST_F(BlobFileStateTest, ForwardIncompatibleCustomField) {
-  SyncPoint::GetInstance()->SetCallBack(
-      "BlobFileState::EncodeTo::CustomFields", [&](void* arg) {
-        std::string* output = static_cast<std::string*>(arg);
-
-        constexpr uint32_t forward_incompatible_tag = (1 << 6) + 1;
-        PutVarint32(output, forward_incompatible_tag);
-
-        PutLengthPrefixedSlice(output, "foobar");
-      });
-  SyncPoint::GetInstance()->EnableProcessing();
-
-  constexpr uint64_t blob_file_number = 456;
-  constexpr uint64_t total_blob_count = 100;
-  constexpr uint64_t total_blob_bytes = 2000000;
-  const std::string checksum_method("CRC32B");
-  const std::string checksum_value("6dbdf23a");
-
-  BlobFileState blob_file_state(blob_file_number, total_blob_count,
-                                total_blob_bytes, checksum_method,
-                                checksum_value);
-
-  std::string encoded;
-  blob_file_state.EncodeTo(&encoded);
-
-  BlobFileState decoded_blob_file_state;
-  Slice input(encoded);
-  const Status s = decoded_blob_file_state.DecodeFrom(&input);
-
-  ASSERT_TRUE(s.IsCorruption());
-  ASSERT_TRUE(std::strstr(s.getState(), "Forward incompatible"));
-
-  SyncPoint::GetInstance()->DisableProcessing();
-  SyncPoint::GetInstance()->ClearAllCallBacks();
-}
-
-}  // namespace ROCKSDB_NAMESPACE
-
-int main(int argc, char** argv) {
-  ::testing::InitGoogleTest(&argc, argv);
-  return RUN_ALL_TESTS();
-}
diff --git a/db/version_edit.cc b/db/version_edit.cc
index 890b5f539..f597aa8d4 100644
--- a/db/version_edit.cc
+++ b/db/version_edit.cc
@@ -57,7 +57,8 @@ enum Tag : uint32_t {
 
   // Forward compatible (aka ignorable) records
   kDbId,
-  kBlobFileState,
+  kBlobFileAddition,
+  kBlobFileGarbage,
 };
 
 enum NewFileCustomTag : uint32_t {
@@ -151,7 +152,8 @@ void VersionEdit::Clear() {
   has_last_sequence_ = false;
   deleted_files_.clear();
   new_files_.clear();
-  blob_file_states_.clear();
+  blob_file_additions_.clear();
+  blob_file_garbages_.clear();
   column_family_ = 0;
   is_column_family_add_ = false;
   is_column_family_drop_ = false;
@@ -276,9 +278,14 @@ bool VersionEdit::EncodeTo(std::string* dst) const {
     PutVarint32(dst, NewFileCustomTag::kTerminate);
   }
 
-  for (const auto& blob_file_state : blob_file_states_) {
-    PutVarint32(dst, kBlobFileState);
-    blob_file_state.EncodeTo(dst);
+  for (const auto& blob_file_addition : blob_file_additions_) {
+    PutVarint32(dst, kBlobFileAddition);
+    blob_file_addition.EncodeTo(dst);
+  }
+
+  for (const auto& blob_file_garbage : blob_file_garbages_) {
+    PutVarint32(dst, kBlobFileGarbage);
+    blob_file_garbage.EncodeTo(dst);
   }
 
   // 0 is default and does not need to be explicitly written
@@ -583,14 +590,25 @@ Status VersionEdit::DecodeFrom(const Slice& src) {
         break;
       }
 
-      case kBlobFileState: {
-        BlobFileState blob_file_state;
-        const Status s = blob_file_state.DecodeFrom(&input);
+      case kBlobFileAddition: {
+        BlobFileAddition blob_file_addition;
+        const Status s = blob_file_addition.DecodeFrom(&input);
         if (!s.ok()) {
           return s;
         }
 
-        blob_file_states_.emplace_back(blob_file_state);
+        blob_file_additions_.emplace_back(blob_file_addition);
+        break;
+      }
+
+      case kBlobFileGarbage: {
+        BlobFileGarbage blob_file_garbage;
+        const Status s = blob_file_garbage.DecodeFrom(&input);
+        if (!s.ok()) {
+          return s;
+        }
+
+        blob_file_garbages_.emplace_back(blob_file_garbage);
         break;
       }
 
@@ -724,9 +742,14 @@ std::string VersionEdit::DebugString(bool hex_key) const {
     r.append(f.file_checksum_func_name);
   }
 
-  for (const auto& blob_file_state : blob_file_states_) {
-    r.append("\n  BlobFileState: ");
-    r.append(blob_file_state.DebugString());
+  for (const auto& blob_file_addition : blob_file_additions_) {
+    r.append("\n  BlobFileAddition: ");
+    r.append(blob_file_addition.DebugString());
+  }
+
+  for (const auto& blob_file_garbage : blob_file_garbages_) {
+    r.append("\n  BlobFileGarbage: ");
+    r.append(blob_file_garbage.DebugString());
   }
 
   r.append("\n  ColumnFamily: ");
@@ -811,14 +834,28 @@ std::string VersionEdit::DebugJSON(int edit_num, bool hex_key) const {
     jw.EndArray();
   }
 
-  if (!blob_file_states_.empty()) {
-    jw << "BlobFileStates";
+  if (!blob_file_additions_.empty()) {
+    jw << "BlobFileAdditions";
+
+    jw.StartArray();
+
+    for (const auto& blob_file_addition : blob_file_additions_) {
+      jw.StartArrayedObject();
+      jw << blob_file_addition;
+      jw.EndArrayedObject();
+    }
+
+    jw.EndArray();
+  }
+
+  if (!blob_file_garbages_.empty()) {
+    jw << "BlobFileGarbages";
 
     jw.StartArray();
 
-    for (const auto& blob_file_state : blob_file_states_) {
+    for (const auto& blob_file_garbage : blob_file_garbages_) {
       jw.StartArrayedObject();
-      jw << blob_file_state;
+      jw << blob_file_garbage;
       jw.EndArrayedObject();
     }
 
diff --git a/db/version_edit.h b/db/version_edit.h
index 88954641d..d08dc4f6a 100644
--- a/db/version_edit.h
+++ b/db/version_edit.h
@@ -13,7 +13,8 @@
 #include <string>
 #include <utility>
 #include <vector>
-#include "db/blob_file_state.h"
+#include "db/blob_file_addition.h"
+#include "db/blob_file_garbage.h"
 #include "db/dbformat.h"
 #include "memory/arena.h"
 #include "rocksdb/cache.h"
@@ -346,25 +347,40 @@ class VersionEdit {
   using NewFiles = std::vector<std::pair<int, FileMetaData>>;
   const NewFiles& GetNewFiles() const { return new_files_; }
 
-  // Add blob file state for the specified file.
-  void AddBlobFileState(uint64_t blob_file_number, uint64_t total_blob_count,
-                        uint64_t total_blob_bytes, uint64_t garbage_blob_count,
-                        uint64_t garbage_blob_bytes,
-                        std::string checksum_method,
-                        std::string checksum_value) {
-    blob_file_states_.emplace_back(
+  // Add a new blob file.
+  void AddBlobFile(uint64_t blob_file_number, uint64_t total_blob_count,
+                   uint64_t total_blob_bytes, std::string checksum_method,
+                   std::string checksum_value) {
+    blob_file_additions_.emplace_back(
         blob_file_number, total_blob_count, total_blob_bytes,
-        garbage_blob_count, garbage_blob_bytes, std::move(checksum_method),
-        std::move(checksum_value));
+        std::move(checksum_method), std::move(checksum_value));
   }
 
-  // Retrieve all the blob file states added.
-  using BlobFileStates = std::vector<BlobFileState>;
-  const BlobFileStates& GetBlobFileStates() const { return blob_file_states_; }
+  // Retrieve all the blob files added.
+  using BlobFileAdditions = std::vector<BlobFileAddition>;
+  const BlobFileAdditions& GetBlobFileAdditions() const {
+    return blob_file_additions_;
+  }
+
+  // Add garbage for an existing blob file.  Note: intentionally broken English
+  // follows.
+  void AddBlobFileGarbage(uint64_t blob_file_number,
+                          uint64_t garbage_blob_count,
+                          uint64_t garbage_blob_bytes) {
+    blob_file_garbages_.emplace_back(blob_file_number, garbage_blob_count,
+                                     garbage_blob_bytes);
+  }
+
+  // Retrieve all the blob file garbage added.
+  using BlobFileGarbages = std::vector<BlobFileGarbage>;
+  const BlobFileGarbages& GetBlobFileGarbages() const {
+    return blob_file_garbages_;
+  }
 
   // Number of edits
   size_t NumEntries() const {
-    return new_files_.size() + deleted_files_.size() + blob_file_states_.size();
+    return new_files_.size() + deleted_files_.size() +
+           blob_file_additions_.size() + blob_file_garbages_.size();
   }
 
   void SetColumnFamily(uint32_t column_family_id) {
@@ -439,7 +455,8 @@ class VersionEdit {
   DeletedFiles deleted_files_;
   NewFiles new_files_;
 
-  BlobFileStates blob_file_states_;
+  BlobFileAdditions blob_file_additions_;
+  BlobFileGarbages blob_file_garbages_;
 
   // Each version edit record should have column_family_ set
   // If it's not set, it is default (0)
diff --git a/db/version_edit_test.cc b/db/version_edit_test.cc
index d03d1d735..056f4adaf 100644
--- a/db/version_edit_test.cc
+++ b/db/version_edit_test.cc
@@ -279,7 +279,7 @@ TEST_F(VersionEditTest, DbId) {
   TestEncodeDecode(edit);
 }
 
-TEST_F(VersionEditTest, BlobFileState) {
+TEST_F(VersionEditTest, BlobFileAdditionAndGarbage) {
   VersionEdit edit;
 
   const std::string checksum_method_prefix = "Hash";
@@ -289,8 +289,6 @@ TEST_F(VersionEditTest, BlobFileState) {
        ++blob_file_number) {
     const uint64_t total_blob_count = blob_file_number << 10;
     const uint64_t total_blob_bytes = blob_file_number << 20;
-    const uint64_t garbage_blob_count = total_blob_count >> 2;
-    const uint64_t garbage_blob_bytes = total_blob_bytes >> 1;
 
     std::string checksum_method(checksum_method_prefix);
     AppendNumberTo(&checksum_method, blob_file_number);
@@ -298,9 +296,14 @@ TEST_F(VersionEditTest, BlobFileState) {
     std::string checksum_value(checksum_value_prefix);
     AppendNumberTo(&checksum_value, blob_file_number);
 
-    edit.AddBlobFileState(blob_file_number, total_blob_count, total_blob_bytes,
-                          garbage_blob_count, garbage_blob_bytes,
-                          checksum_method, checksum_value);
+    edit.AddBlobFile(blob_file_number, total_blob_count, total_blob_bytes,
+                     checksum_method, checksum_value);
+
+    const uint64_t garbage_blob_count = total_blob_count >> 2;
+    const uint64_t garbage_blob_bytes = total_blob_bytes >> 1;
+
+    edit.AddBlobFileGarbage(blob_file_number, garbage_blob_count,
+                            garbage_blob_bytes);
   }
 
   TestEncodeDecode(edit);
diff --git a/src.mk b/src.mk
index 9e9dd1a9e..a85fe60d2 100644
--- a/src.mk
+++ b/src.mk
@@ -4,7 +4,8 @@ LIB_SOURCES =                                                   \
   cache/lru_cache.cc                                            \
   cache/sharded_cache.cc                                        \
   db/arena_wrapped_db_iter.cc                                   \
-  db/blob_file_state.cc                                         \
+  db/blob_file_addition.cc                                      \
+  db/blob_file_garbage.cc                                       \
   db/builder.cc                                                 \
   db/c.cc                                                       \
   db/column_family.cc                                           \
@@ -297,7 +298,8 @@ MAIN_SOURCES =                                                          \
   cache/cache_bench.cc                                                  \
   cache/cache_test.cc                                                   \
   db_stress_tool/db_stress.cc                                           \
-  db/blob_file_state_test.cc                                            \
+  db/blob_file_addition_test.cc                                         \
+  db/blob_file_garbage_test.cc                                          \
   db/column_family_test.cc                                              \
   db/compact_files_test.cc                                              \
   db/compaction/compaction_iterator_test.cc                             \