Clean up blob files based on the linked SST set (#7001)

Summary:
The earlier `VersionBuilder` code only cleaned up blob files that were
marked as entirely consisting of garbage using `VersionEdits` with
`BlobFileGarbage`. This covers the cases when table files go through
regular compaction, where we iterate through the KVs and thus have an
opportunity to calculate the amount of garbage (that is, most cases).
However, it does not help when table files are simply dropped (e.g. deletion
compactions or the `DeleteFile` API). To deal with such cases, the patch
adds logic that cleans up all blob files at the head of the list until the first
one with linked SSTs is found. (As an example, let's assume we have blob files
with numbers 1..10, and the first one with any linked SSTs is number 8.
This means that SSTs in the `Version` only rely on blob files with numbers >= 8,
and thus 1..7 are no longer needed.)

The code change itself is pretty small; however, changing the logic like this
necessitated changes to some tests that have been added recently (namely
to the ones that use blob files in isolation, i.e. without any table files referring
to them). Some of these cases were fixed by bypassing `VersionBuilder` altogether
in order to keep the tests simple (which actually makes them more proper unit tests
as well), while the `VersionBuilder` unit tests were fixed by adding dummy table
files to the test cases as needed.
Pull Request resolved: https://github.com/facebook/rocksdb/pull/7001

Test Plan: `make check`

Reviewed By: riversand963

Differential Revision: D22119474

Pulled By: ltamasi

fbshipit-source-id: c6547141355667d4291d9661d6518eb741e7b54a

db/db_test.cc

@@ -2344,24 +2344,31 @@ TEST_F(DBTest, GetLiveBlobFiles) {
   ColumnFamilyData* const cfd = versions->GetColumnFamilySet()->GetDefault();
   assert(cfd);
 
-  // Add a live blob file.
-  VersionEdit edit;
+  Version* const version = cfd->current();
+  assert(version);
+
+  VersionStorageInfo* const storage_info = version->storage_info();
+  assert(storage_info);
 
+  // Add a live blob file.
   constexpr uint64_t blob_file_number = 234;
   constexpr uint64_t total_blob_count = 555;
   constexpr uint64_t total_blob_bytes = 66666;
   constexpr char checksum_method[] = "CRC32";
   constexpr char checksum_value[] = "3d87ff57";
 
-  edit.AddBlobFile(blob_file_number, total_blob_count, total_blob_bytes,
-                   checksum_method, checksum_value);
+  auto shared_meta = SharedBlobFileMetaData::Create(
+      blob_file_number, total_blob_count, total_blob_bytes, checksum_method,
+      checksum_value);
 
-  dbfull()->TEST_LockMutex();
-  Status s = versions->LogAndApply(cfd, *cfd->GetLatestMutableCFOptions(),
-                                   &edit, dbfull()->mutex());
-  dbfull()->TEST_UnlockMutex();
+  constexpr uint64_t garbage_blob_count = 0;
+  constexpr uint64_t garbage_blob_bytes = 0;
 
-  ASSERT_OK(s);
+  auto meta = BlobFileMetaData::Create(std::move(shared_meta),
+                                       BlobFileMetaData::LinkedSsts(),
+                                       garbage_blob_count, garbage_blob_bytes);
+
+  storage_info->AddBlobFile(std::move(meta));
 
   // Make sure it appears in the results returned by GetLiveFiles.
   uint64_t manifest_size = 0;

db/obsolete_files_test.cc

@@ -202,37 +202,41 @@ TEST_F(ObsoleteFilesTest, BlobFiles) {
   ColumnFamilyData* const cfd = versions->GetColumnFamilySet()->GetDefault();
   assert(cfd);
 
-  // Add a blob file that consists of nothing but garbage (and is thus obsolete)
-  // and another one that is live.
-  VersionEdit edit;
+  const ImmutableCFOptions* const ioptions = cfd->ioptions();
+  assert(ioptions);
+  assert(!ioptions->cf_paths.empty());
 
+  const std::string& path = ioptions->cf_paths.front().path;
+
+  // Add an obsolete blob file.
   constexpr uint64_t first_blob_file_number = 234;
-  constexpr uint64_t first_total_blob_count = 555;
-  constexpr uint64_t first_total_blob_bytes = 66666;
-  constexpr char first_checksum_method[] = "CRC32";
-  constexpr char first_checksum_value[] = "3d87ff57";
+  versions->AddObsoleteBlobFile(first_blob_file_number, path);
+
+  // Add a live blob file.
+  Version* const version = cfd->current();
+  assert(version);
 
-  edit.AddBlobFile(first_blob_file_number, first_total_blob_count,
-                   first_total_blob_bytes, first_checksum_method,
-                   first_checksum_value);
-  edit.AddBlobFileGarbage(first_blob_file_number, first_total_blob_count,
-                          first_total_blob_bytes);
+  VersionStorageInfo* const storage_info = version->storage_info();
+  assert(storage_info);
 
   constexpr uint64_t second_blob_file_number = 456;
   constexpr uint64_t second_total_blob_count = 100;
   constexpr uint64_t second_total_blob_bytes = 2000000;
   constexpr char second_checksum_method[] = "CRC32B";
   constexpr char second_checksum_value[] = "6dbdf23a";
-  edit.AddBlobFile(second_blob_file_number, second_total_blob_count,
-                   second_total_blob_bytes, second_checksum_method,
-                   second_checksum_value);
 
-  dbfull()->TEST_LockMutex();
-  Status s = versions->LogAndApply(cfd, *cfd->GetLatestMutableCFOptions(),
-                                   &edit, dbfull()->mutex());
-  dbfull()->TEST_UnlockMutex();
+  auto shared_meta = SharedBlobFileMetaData::Create(
+      second_blob_file_number, second_total_blob_count, second_total_blob_bytes,
+      second_checksum_method, second_checksum_value);
 
-  ASSERT_OK(s);
+  constexpr uint64_t second_garbage_blob_count = 0;
+  constexpr uint64_t second_garbage_blob_bytes = 0;
+
+  auto meta = BlobFileMetaData::Create(
+      std::move(shared_meta), BlobFileMetaData::LinkedSsts(),
+      second_garbage_blob_count, second_garbage_blob_bytes);
+
+  storage_info->AddBlobFile(std::move(meta));
 
   // Check for obsolete files and make sure the first blob file is picked up
   // and grabbed for purge. The second blob file should be on the live list.
@@ -261,10 +265,6 @@ TEST_F(ObsoleteFilesTest, BlobFiles) {
   // list and adjusting the pending file number. We add the two files
   // above as well as two additional ones, where one is old
   // and should be cleaned up, and the other is still pending.
-  assert(cfd->ioptions());
-  assert(!cfd->ioptions()->cf_paths.empty());
-  const std::string& path = cfd->ioptions()->cf_paths.front().path;
-
   constexpr uint64_t old_blob_file_number = 123;
   constexpr uint64_t pending_blob_file_number = 567;
 

db/version_builder.cc

@@ -722,16 +722,26 @@ class VersionBuilder::Rep {
     return meta;
   }
 
-  void AddBlobFileIfNeeded(
-      VersionStorageInfo* vstorage,
-      const std::shared_ptr<BlobFileMetaData>& meta) const {
+  // Add the blob file specified by meta to *vstorage if it is determined to
+  // contain valid data (blobs). We make this decision based on the amount
+  // of garbage in the file, and whether the file or any lower-numbered blob
+  // files have any linked SSTs. The latter condition is tracked using the
+  // flag *found_first_non_empty.
+  void AddBlobFileIfNeeded(VersionStorageInfo* vstorage,
+                           const std::shared_ptr<BlobFileMetaData>& meta,
+                           bool* found_first_non_empty) const {
     assert(vstorage);
     assert(meta);
+    assert(found_first_non_empty);
 
-    if (meta->GetGarbageBlobCount() < meta->GetTotalBlobCount() ||
-        !meta->GetLinkedSsts().empty()) {
-      vstorage->AddBlobFile(meta);
+    if (!meta->GetLinkedSsts().empty()) {
+      (*found_first_non_empty) = true;
+    } else if (!(*found_first_non_empty) ||
+               meta->GetGarbageBlobCount() >= meta->GetTotalBlobCount()) {
+      return;
     }
+
+    vstorage->AddBlobFile(meta);
   }
 
   // Merge the blob file metadata from the base version with the changes (edits)
@@ -740,6 +750,8 @@ class VersionBuilder::Rep {
     assert(base_vstorage_);
     assert(vstorage);
 
+    bool found_first_non_empty = false;
+
     const auto& base_blob_files = base_vstorage_->GetBlobFiles();
     auto base_it = base_blob_files.begin();
     const auto base_it_end = base_blob_files.end();
@@ -753,11 +765,8 @@ class VersionBuilder::Rep {
 
       if (base_blob_file_number < delta_blob_file_number) {
         const auto& base_meta = base_it->second;
-        assert(base_meta);
-        assert(base_meta->GetGarbageBlobCount() <
-               base_meta->GetTotalBlobCount());
 
-        vstorage->AddBlobFile(base_meta);
+        AddBlobFileIfNeeded(vstorage, base_meta, &found_first_non_empty);
 
         ++base_it;
       } else if (delta_blob_file_number < base_blob_file_number) {
@@ -775,7 +784,7 @@ class VersionBuilder::Rep {
 
         auto meta = GetOrCreateMetaDataForExistingBlobFile(base_meta, delta);
 
-        AddBlobFileIfNeeded(vstorage, meta);
+        AddBlobFileIfNeeded(vstorage, meta, &found_first_non_empty);
 
         ++base_it;
         ++delta_it;
@@ -784,10 +793,9 @@ class VersionBuilder::Rep {
 
     while (base_it != base_it_end) {
       const auto& base_meta = base_it->second;
-      assert(base_meta);
-      assert(base_meta->GetGarbageBlobCount() < base_meta->GetTotalBlobCount());
 
-      vstorage->AddBlobFile(base_meta);
+      AddBlobFileIfNeeded(vstorage, base_meta, &found_first_non_empty);
+
       ++base_it;
     }
 
@@ -796,7 +804,7 @@ class VersionBuilder::Rep {
 
       auto meta = CreateMetaDataForNewBlobFile(delta);
 
-      AddBlobFileIfNeeded(vstorage, meta);
+      AddBlobFileIfNeeded(vstorage, meta, &found_first_non_empty);
 
       ++delta_it;
     }

db/version_builder_test.cc

@@ -96,6 +96,44 @@ class VersionBuilderTest : public testing::Test {
     vstorage_.AddBlobFile(std::move(meta));
   }
 
+  void AddDummyFile(uint64_t table_file_number, uint64_t blob_file_number) {
+    constexpr int level = 0;
+    constexpr char smallest[] = "bar";
+    constexpr char largest[] = "foo";
+    constexpr uint64_t file_size = 100;
+    constexpr uint32_t path_id = 0;
+    constexpr SequenceNumber smallest_seq = 0;
+    constexpr SequenceNumber largest_seq = 0;
+    constexpr uint64_t num_entries = 0;
+    constexpr uint64_t num_deletions = 0;
+    constexpr bool sampled = false;
+
+    Add(level, table_file_number, smallest, largest, file_size, path_id,
+        smallest_seq, largest_seq, num_entries, num_deletions, sampled,
+        smallest_seq, largest_seq, blob_file_number);
+  }
+
+  void AddDummyFileToEdit(VersionEdit* edit, uint64_t table_file_number,
+                          uint64_t blob_file_number) {
+    assert(edit);
+
+    constexpr int level = 0;
+    constexpr uint32_t path_id = 0;
+    constexpr uint64_t file_size = 100;
+    constexpr char smallest[] = "bar";
+    constexpr char largest[] = "foo";
+    constexpr SequenceNumber smallest_seqno = 100;
+    constexpr SequenceNumber largest_seqno = 300;
+    constexpr bool marked_for_compaction = false;
+
+    edit->AddFile(level, table_file_number, path_id, file_size,
+                  GetInternalKey(smallest), GetInternalKey(largest),
+                  smallest_seqno, largest_seqno, marked_for_compaction,
+                  blob_file_number, kUnknownOldestAncesterTime,
+                  kUnknownFileCreationTime, kUnknownFileChecksum,
+                  kUnknownFileChecksumFuncName);
+  }
+
   static std::shared_ptr<BlobFileMetaData> GetBlobFileMetaData(
       const VersionStorageInfo::BlobFiles& blob_files,
       uint64_t blob_file_number) {
@@ -627,6 +665,10 @@ TEST_F(VersionBuilderTest, ApplyBlobFileAddition) {
   edit.AddBlobFile(blob_file_number, total_blob_count, total_blob_bytes,
                    checksum_method, checksum_value);
 
+  // Add dummy table file to ensure the blob file is referenced.
+  constexpr uint64_t table_file_number = 1;
+  AddDummyFileToEdit(&edit, table_file_number, blob_file_number);
+
   ASSERT_OK(builder.Apply(&edit));
 
   constexpr bool force_consistency_checks = false;
@@ -647,9 +689,12 @@ TEST_F(VersionBuilderTest, ApplyBlobFileAddition) {
   ASSERT_EQ(new_meta->GetTotalBlobBytes(), total_blob_bytes);
   ASSERT_EQ(new_meta->GetChecksumMethod(), checksum_method);
   ASSERT_EQ(new_meta->GetChecksumValue(), checksum_value);
-  ASSERT_TRUE(new_meta->GetLinkedSsts().empty());
+  ASSERT_EQ(new_meta->GetLinkedSsts(),
+            BlobFileMetaData::LinkedSsts{table_file_number});
   ASSERT_EQ(new_meta->GetGarbageBlobCount(), 0);
   ASSERT_EQ(new_meta->GetGarbageBlobBytes(), 0);
+
+  UnrefFilesInVersion(&new_vstorage);
 }
 
 TEST_F(VersionBuilderTest, ApplyBlobFileAdditionAlreadyInBase) {
@@ -715,6 +760,7 @@ TEST_F(VersionBuilderTest, ApplyBlobFileAdditionAlreadyApplied) {
 TEST_F(VersionBuilderTest, ApplyBlobFileGarbageFileInBase) {
   // Increase the amount of garbage for a blob file present in the base version.
 
+  constexpr uint64_t table_file_number = 1;
   constexpr uint64_t blob_file_number = 1234;
   constexpr uint64_t total_blob_count = 5678;
   constexpr uint64_t total_blob_bytes = 999999;
@@ -724,13 +770,16 @@ TEST_F(VersionBuilderTest, ApplyBlobFileGarbageFileInBase) {
   constexpr uint64_t garbage_blob_bytes = 456789;
 
   AddBlob(blob_file_number, total_blob_count, total_blob_bytes, checksum_method,
-          checksum_value, BlobFileMetaData::LinkedSsts(), garbage_blob_count,
-          garbage_blob_bytes);
+          checksum_value, BlobFileMetaData::LinkedSsts{table_file_number},
+          garbage_blob_count, garbage_blob_bytes);
 
   const auto meta =
       GetBlobFileMetaData(vstorage_.GetBlobFiles(), blob_file_number);
   ASSERT_NE(meta, nullptr);
 
+  // Add dummy table file to ensure the blob file is referenced.
+  AddDummyFile(table_file_number, blob_file_number);
+
   EnvOptions env_options;
   constexpr TableCache* table_cache = nullptr;
   constexpr VersionSet* version_set = nullptr;
@@ -767,11 +816,14 @@ TEST_F(VersionBuilderTest, ApplyBlobFileGarbageFileInBase) {
   ASSERT_EQ(new_meta->GetTotalBlobBytes(), total_blob_bytes);
   ASSERT_EQ(new_meta->GetChecksumMethod(), checksum_method);
   ASSERT_EQ(new_meta->GetChecksumValue(), checksum_value);
-  ASSERT_TRUE(new_meta->GetLinkedSsts().empty());
+  ASSERT_EQ(new_meta->GetLinkedSsts(),
+            BlobFileMetaData::LinkedSsts{table_file_number});
   ASSERT_EQ(new_meta->GetGarbageBlobCount(),
             garbage_blob_count + new_garbage_blob_count);
   ASSERT_EQ(new_meta->GetGarbageBlobBytes(),
             garbage_blob_bytes + new_garbage_blob_bytes);
+
+  UnrefFilesInVersion(&new_vstorage);
 }
 
 TEST_F(VersionBuilderTest, ApplyBlobFileGarbageFileAdditionApplied) {
@@ -795,6 +847,10 @@ TEST_F(VersionBuilderTest, ApplyBlobFileGarbageFileAdditionApplied) {
   addition.AddBlobFile(blob_file_number, total_blob_count, total_blob_bytes,
                        checksum_method, checksum_value);
 
+  // Add dummy table file to ensure the blob file is referenced.
+  constexpr uint64_t table_file_number = 1;
+  AddDummyFileToEdit(&addition, table_file_number, blob_file_number);
+
   ASSERT_OK(builder.Apply(&addition));
 
   constexpr uint64_t garbage_blob_count = 123;
@@ -825,9 +881,12 @@ TEST_F(VersionBuilderTest, ApplyBlobFileGarbageFileAdditionApplied) {
   ASSERT_EQ(new_meta->GetTotalBlobBytes(), total_blob_bytes);
   ASSERT_EQ(new_meta->GetChecksumMethod(), checksum_method);
   ASSERT_EQ(new_meta->GetChecksumValue(), checksum_value);
-  ASSERT_TRUE(new_meta->GetLinkedSsts().empty());
+  ASSERT_EQ(new_meta->GetLinkedSsts(),
+            BlobFileMetaData::LinkedSsts{table_file_number});
   ASSERT_EQ(new_meta->GetGarbageBlobCount(), garbage_blob_count);
   ASSERT_EQ(new_meta->GetGarbageBlobBytes(), garbage_blob_bytes);
+
+  UnrefFilesInVersion(&new_vstorage);
 }
 
 TEST_F(VersionBuilderTest, ApplyBlobFileGarbageFileNotFound) {
@@ -857,7 +916,8 @@ TEST_F(VersionBuilderTest, ApplyBlobFileGarbageFileNotFound) {
 
 TEST_F(VersionBuilderTest, SaveBlobFilesTo) {
   // Add three blob files to base version.
-  for (uint64_t i = 1; i <= 3; ++i) {
+  for (uint64_t i = 3; i >= 1; --i) {
+    const uint64_t table_file_number = i;
     const uint64_t blob_file_number = i;
     const uint64_t total_blob_count = i * 1000;
     const uint64_t total_blob_bytes = i * 1000000;
@@ -866,8 +926,12 @@ TEST_F(VersionBuilderTest, SaveBlobFilesTo) {
 
     AddBlob(blob_file_number, total_blob_count, total_blob_bytes,
             /* checksum_method */ std::string(),
-            /* checksum_value */ std::string(), BlobFileMetaData::LinkedSsts(),
-            garbage_blob_count, garbage_blob_bytes);
+            /* checksum_value */ std::string(),
+            BlobFileMetaData::LinkedSsts{table_file_number}, garbage_blob_count,
+            garbage_blob_bytes);
+
+    // Add dummy table file to ensure the blob file is referenced.
+    AddDummyFile(table_file_number, blob_file_number);
   }
 
   EnvOptions env_options;
@@ -882,13 +946,15 @@ TEST_F(VersionBuilderTest, SaveBlobFilesTo) {
   // Add some garbage to the second and third blob files. The second blob file
   // remains valid since it does not consist entirely of garbage yet. The third
   // blob file is all garbage after the edit and will not be part of the new
-  // version.
+  // version. The corresponding dummy table file is also removed for
+  // consistency.
   edit.AddBlobFileGarbage(/* blob_file_number */ 2,
                           /* garbage_blob_count */ 200,
                           /* garbage_blob_bytes */ 100000);
   edit.AddBlobFileGarbage(/* blob_file_number */ 3,
                           /* garbage_blob_count */ 2700,
                           /* garbage_blob_bytes */ 2940000);
+  edit.DeleteFile(/* level */ 0, /* file_number */ 3);
 
   // Add a fourth blob file.
   edit.AddBlobFile(/* blob_file_number */ 4, /* total_blob_count */ 4000,
@@ -934,6 +1000,32 @@ TEST_F(VersionBuilderTest, SaveBlobFilesTo) {
   ASSERT_EQ(meta4->GetTotalBlobBytes(), 4000000);
   ASSERT_EQ(meta4->GetGarbageBlobCount(), 0);
   ASSERT_EQ(meta4->GetGarbageBlobBytes(), 0);
+
+  // Delete the first table file, which makes the first blob file obsolete
+  // since it's at the head and unreferenced.
+  VersionBuilder second_builder(env_options, &ioptions_, table_cache,
+                                &new_vstorage, version_set);
+
+  VersionEdit second_edit;
+  second_edit.DeleteFile(/* level */ 0, /* file_number */ 1);
+
+  ASSERT_OK(second_builder.Apply(&second_edit));
+
+  VersionStorageInfo newer_vstorage(&icmp_, ucmp_, options_.num_levels,
+                                    kCompactionStyleLevel, &new_vstorage,
+                                    force_consistency_checks);
+
+  ASSERT_OK(second_builder.SaveTo(&newer_vstorage));
+
+  const auto& newer_blob_files = newer_vstorage.GetBlobFiles();
+  ASSERT_EQ(newer_blob_files.size(), 2);
+
+  const auto newer_meta1 = GetBlobFileMetaData(newer_blob_files, 1);
+
+  ASSERT_EQ(newer_meta1, nullptr);
+
+  UnrefFilesInVersion(&newer_vstorage);
+  UnrefFilesInVersion(&new_vstorage);
 }
 
 TEST_F(VersionBuilderTest, CheckConsistencyForBlobFiles) {

db/version_set.h

@@ -1161,6 +1161,18 @@ class VersionSet {
   // Get the IO Status returned by written Manifest.
   const IOStatus& io_status() const { return io_status_; }
 
+  void TEST_CreateAndAppendVersion(ColumnFamilyData* cfd) {
+    assert(cfd);
+
+    const auto& mutable_cf_options = *cfd->GetLatestMutableCFOptions();
+    Version* const version =
+        new Version(cfd, this, file_options_, mutable_cf_options);
+
+    constexpr bool update_stats = false;
+    version->PrepareApply(mutable_cf_options, update_stats);
+    AppendVersion(cfd, version);
+  }
+
  protected:
   using VersionBuilderMap =
       std::unordered_map<uint32_t,

db/version_set_test.cc

@@ -911,7 +911,17 @@ TEST_F(VersionSetTest, PersistBlobFileStateInNewManifest) {
   // garbage in it, and one without any garbage.
   NewDB();
 
-  VersionEdit edit;
+  assert(versions_);
+  assert(versions_->GetColumnFamilySet());
+
+  ColumnFamilyData* const cfd = versions_->GetColumnFamilySet()->GetDefault();
+  assert(cfd);
+
+  Version* const version = cfd->current();
+  assert(version);
+
+  VersionStorageInfo* const storage_info = version->storage_info();
+  assert(storage_info);
 
   {
     constexpr uint64_t blob_file_number = 123;
@@ -920,13 +930,19 @@ TEST_F(VersionSetTest, PersistBlobFileStateInNewManifest) {
     constexpr char checksum_method[] = "SHA1";
     constexpr char checksum_value[] =
         "bdb7f34a59dfa1592ce7f52e99f98c570c525cbd";
+
+    auto shared_meta = SharedBlobFileMetaData::Create(
+        blob_file_number, total_blob_count, total_blob_bytes, checksum_method,
+        checksum_value);
+
     constexpr uint64_t garbage_blob_count = 89;
     constexpr uint64_t garbage_blob_bytes = 1000000;
 
-    edit.AddBlobFile(blob_file_number, total_blob_count, total_blob_bytes,
-                     checksum_method, checksum_value);
-    edit.AddBlobFileGarbage(blob_file_number, garbage_blob_count,
-                            garbage_blob_bytes);
+    auto meta = BlobFileMetaData::Create(
+        std::move(shared_meta), BlobFileMetaData::LinkedSsts(),
+        garbage_blob_count, garbage_blob_bytes);
+
+    storage_info->AddBlobFile(std::move(meta));
   }
 
   {
@@ -936,20 +952,19 @@ TEST_F(VersionSetTest, PersistBlobFileStateInNewManifest) {
     constexpr char checksum_method[] = "CRC32";
     constexpr char checksum_value[] = "3d87ff57";
 
-    edit.AddBlobFile(blob_file_number, total_blob_count, total_blob_bytes,
-                     checksum_method, checksum_value);
-  }
+    auto shared_meta = SharedBlobFileMetaData::Create(
+        blob_file_number, total_blob_count, total_blob_bytes, checksum_method,
+        checksum_value);
 
-  assert(versions_);
-  assert(versions_->GetColumnFamilySet());
+    constexpr uint64_t garbage_blob_count = 0;
+    constexpr uint64_t garbage_blob_bytes = 0;
 
-  mutex_.Lock();
-  Status s =
-      versions_->LogAndApply(versions_->GetColumnFamilySet()->GetDefault(),
-                             mutable_cf_options_, &edit, &mutex_);
-  mutex_.Unlock();
+    auto meta = BlobFileMetaData::Create(
+        std::move(shared_meta), BlobFileMetaData::LinkedSsts(),
+        garbage_blob_count, garbage_blob_bytes);
 
-  ASSERT_OK(s);
+    storage_info->AddBlobFile(std::move(meta));
+  }
 
   // Force the creation of a new manifest file and make sure metadata for
   // the blob files is re-persisted.
@@ -969,9 +984,9 @@ TEST_F(VersionSetTest, PersistBlobFileStateInNewManifest) {
   mutex_.Lock();
   constexpr FSDirectory* db_directory = nullptr;
   constexpr bool new_descriptor_log = true;
-  s = versions_->LogAndApply(versions_->GetColumnFamilySet()->GetDefault(),
-                             mutable_cf_options_, &dummy, &mutex_, db_directory,
-                             new_descriptor_log);
+  Status s = versions_->LogAndApply(
+      versions_->GetColumnFamilySet()->GetDefault(), mutable_cf_options_,
+      &dummy, &mutex_, db_directory, new_descriptor_log);
   mutex_.Unlock();
 
   ASSERT_OK(s);
@@ -992,7 +1007,11 @@ TEST_F(VersionSetTest, AddLiveBlobFiles) {
   ColumnFamilyData* const cfd = versions_->GetColumnFamilySet()->GetDefault();
   assert(cfd);
 
-  VersionEdit first;
+  Version* const first_version = cfd->current();
+  assert(first_version);
+
+  VersionStorageInfo* const first_storage_info = first_version->storage_info();
+  assert(first_storage_info);
 
   constexpr uint64_t first_blob_file_number = 234;
   constexpr uint64_t first_total_blob_count = 555;
@@ -1000,49 +1019,59 @@ TEST_F(VersionSetTest, AddLiveBlobFiles) {
   constexpr char first_checksum_method[] = "CRC32";
   constexpr char first_checksum_value[] = "3d87ff57";
 
-  first.AddBlobFile(first_blob_file_number, first_total_blob_count,
-                    first_total_blob_bytes, first_checksum_method,
-                    first_checksum_value);
+  auto first_shared_meta = SharedBlobFileMetaData::Create(
+      first_blob_file_number, first_total_blob_count, first_total_blob_bytes,
+      first_checksum_method, first_checksum_value);
 
-  mutex_.Lock();
-  Status s = versions_->LogAndApply(cfd, mutable_cf_options_, &first, &mutex_);
-  mutex_.Unlock();
+  constexpr uint64_t garbage_blob_count = 0;
+  constexpr uint64_t garbage_blob_bytes = 0;
 
-  ASSERT_OK(s);
+  auto first_meta = BlobFileMetaData::Create(
+      std::move(first_shared_meta), BlobFileMetaData::LinkedSsts(),
+      garbage_blob_count, garbage_blob_bytes);
+
+  first_storage_info->AddBlobFile(first_meta);
 
   // Reference the version so it stays alive even after the following version
   // edit.
-  Version* const version = cfd->current();
-  assert(version);
-
-  version->Ref();
+  first_version->Ref();
 
   // Get live files directly from version.
   std::vector<uint64_t> version_table_files;
   std::vector<uint64_t> version_blob_files;
 
-  version->AddLiveFiles(&version_table_files, &version_blob_files);
+  first_version->AddLiveFiles(&version_table_files, &version_blob_files);
 
   ASSERT_EQ(version_blob_files.size(), 1);
   ASSERT_EQ(version_blob_files[0], first_blob_file_number);
 
-  // Add another blob file.
-  VersionEdit second;
+  // Create a new version containing an additional blob file.
+  versions_->TEST_CreateAndAppendVersion(cfd);
+
+  Version* const second_version = cfd->current();
+  assert(second_version);
+  assert(second_version != first_version);
+
+  VersionStorageInfo* const second_storage_info =
+      second_version->storage_info();
+  assert(second_storage_info);
 
   constexpr uint64_t second_blob_file_number = 456;
   constexpr uint64_t second_total_blob_count = 100;
   constexpr uint64_t second_total_blob_bytes = 2000000;
   constexpr char second_checksum_method[] = "CRC32B";
   constexpr char second_checksum_value[] = "6dbdf23a";
-  second.AddBlobFile(second_blob_file_number, second_total_blob_count,
-                     second_total_blob_bytes, second_checksum_method,
-                     second_checksum_value);
 
-  mutex_.Lock();
-  s = versions_->LogAndApply(cfd, mutable_cf_options_, &second, &mutex_);
-  mutex_.Unlock();
+  auto second_shared_meta = SharedBlobFileMetaData::Create(
+      second_blob_file_number, second_total_blob_count, second_total_blob_bytes,
+      second_checksum_method, second_checksum_value);
 
-  ASSERT_OK(s);
+  auto second_meta = BlobFileMetaData::Create(
+      std::move(second_shared_meta), BlobFileMetaData::LinkedSsts(),
+      garbage_blob_count, garbage_blob_bytes);
+
+  second_storage_info->AddBlobFile(std::move(first_meta));
+  second_storage_info->AddBlobFile(std::move(second_meta));
 
   // Get all live files from version set. Note that the result contains
   // duplicates.
@@ -1057,14 +1086,15 @@ TEST_F(VersionSetTest, AddLiveBlobFiles) {
   ASSERT_EQ(all_blob_files[2], second_blob_file_number);
 
   // Clean up previous version.
-  version->Unref();
+  first_version->Unref();
 }
 
 TEST_F(VersionSetTest, ObsoleteBlobFile) {
-  // Initialize the database and add a blob file (with no garbage just yet).
+  // Initialize the database and add a blob file that is entirely garbage
+  // and thus can immediately be marked obsolete.
   NewDB();
 
-  VersionEdit addition;
+  VersionEdit edit;
 
   constexpr uint64_t blob_file_number = 234;
   constexpr uint64_t total_blob_count = 555;
@@ -1072,29 +1102,15 @@ TEST_F(VersionSetTest, ObsoleteBlobFile) {
   constexpr char checksum_method[] = "CRC32";
   constexpr char checksum_value[] = "3d87ff57";
 
-  addition.AddBlobFile(blob_file_number, total_blob_count, total_blob_bytes,
-                       checksum_method, checksum_value);
+  edit.AddBlobFile(blob_file_number, total_blob_count, total_blob_bytes,
+                   checksum_method, checksum_value);
 
-  assert(versions_);
-  assert(versions_->GetColumnFamilySet());
+  edit.AddBlobFileGarbage(blob_file_number, total_blob_count, total_blob_bytes);
 
   mutex_.Lock();
   Status s =
       versions_->LogAndApply(versions_->GetColumnFamilySet()->GetDefault(),
-                             mutable_cf_options_, &addition, &mutex_);
-  mutex_.Unlock();
-
-  ASSERT_OK(s);
-
-  // Mark the entire blob file garbage.
-  VersionEdit garbage;
-
-  garbage.AddBlobFileGarbage(blob_file_number, total_blob_count,
-                             total_blob_bytes);
-
-  mutex_.Lock();
-  s = versions_->LogAndApply(versions_->GetColumnFamilySet()->GetDefault(),
-                             mutable_cf_options_, &garbage, &mutex_);
+                             mutable_cf_options_, &edit, &mutex_);
   mutex_.Unlock();
 
   ASSERT_OK(s);