Maintain the set of linked SSTs in BlobFileMetaData (#6945)

Summary:
The `FileMetaData` objects associated with table files already contain the
number of the oldest blob file referenced by the SST in question. This patch
adds the inverse mapping to `BlobFileMetaData`, namely the set of table file
numbers for which the oldest blob file link points to the given blob file (these
are referred to as *linked SSTs*). This mapping will be used by the GC logic.

Implementation-wise, the patch builds on the `BlobFileMetaDataDelta`
functionality introduced in https://github.com/facebook/rocksdb/pull/6835: newly linked/unlinked SSTs are
accumulated in `BlobFileMetaDataDelta`, and the changes to the linked SST set
are applied in one shot when the new `Version` is saved. The patch also reworks
the blob file related consistency checks in `VersionBuilder` so they validate the
consistency of the forward table file -> blob file links and the backward blob file ->
table file links for blob files that are part of the `Version`.
Pull Request resolved: https://github.com/facebook/rocksdb/pull/6945

Test Plan: `make check`

Reviewed By: riversand963

Differential Revision: D21912228

Pulled By: ltamasi

fbshipit-source-id: c5bc7acf6e729a8fccbb12672dd5cd00f6f000f8

db/blob/blob_file_meta.cc

@@ -38,8 +38,15 @@ std::string BlobFileMetaData::DebugString() const {
 std::ostream& operator<<(std::ostream& os, const BlobFileMetaData& meta) {
   const auto& shared_meta = meta.GetSharedMeta();
   assert(shared_meta);
+  os << (*shared_meta);
 
-  os << (*shared_meta) << " garbage_blob_count: " << meta.GetGarbageBlobCount()
+  os << " linked_ssts: {";
+  for (uint64_t file_number : meta.GetLinkedSsts()) {
+    os << ' ' << file_number;
+  }
+  os << " }";
+
+  os << " garbage_blob_count: " << meta.GetGarbageBlobCount()
      << " garbage_blob_bytes: " << meta.GetGarbageBlobBytes();
 
   return os;

db/blob/blob_file_meta.h

@@ -5,12 +5,13 @@
 
 #pragma once
 
-#include "rocksdb/rocksdb_namespace.h"
-
 #include <cassert>
 #include <iosfwd>
 #include <memory>
 #include <string>
+#include <unordered_set>
+
+#include "rocksdb/rocksdb_namespace.h"
 
 namespace ROCKSDB_NAMESPACE {
 
@@ -90,11 +91,15 @@ std::ostream& operator<<(std::ostream& os,
 
 class BlobFileMetaData {
  public:
+  using LinkedSsts = std::unordered_set<uint64_t>;
+
   static std::shared_ptr<BlobFileMetaData> Create(
       std::shared_ptr<SharedBlobFileMetaData> shared_meta,
-      uint64_t garbage_blob_count, uint64_t garbage_blob_bytes) {
-    return std::shared_ptr<BlobFileMetaData>(new BlobFileMetaData(
-        std::move(shared_meta), garbage_blob_count, garbage_blob_bytes));
+      LinkedSsts linked_ssts, uint64_t garbage_blob_count,
+      uint64_t garbage_blob_bytes) {
+    return std::shared_ptr<BlobFileMetaData>(
+        new BlobFileMetaData(std::move(shared_meta), std::move(linked_ssts),
+                             garbage_blob_count, garbage_blob_bytes));
   }
 
   BlobFileMetaData(const BlobFileMetaData&) = delete;
@@ -128,6 +133,8 @@ class BlobFileMetaData {
     return shared_meta_->GetChecksumValue();
   }
 
+  const LinkedSsts& GetLinkedSsts() const { return linked_ssts_; }
+
   uint64_t GetGarbageBlobCount() const { return garbage_blob_count_; }
   uint64_t GetGarbageBlobBytes() const { return garbage_blob_bytes_; }
 
@@ -135,8 +142,10 @@ class BlobFileMetaData {
 
  private:
   BlobFileMetaData(std::shared_ptr<SharedBlobFileMetaData> shared_meta,
-                   uint64_t garbage_blob_count, uint64_t garbage_blob_bytes)
+                   LinkedSsts linked_ssts, uint64_t garbage_blob_count,
+                   uint64_t garbage_blob_bytes)
       : shared_meta_(std::move(shared_meta)),
+        linked_ssts_(std::move(linked_ssts)),
         garbage_blob_count_(garbage_blob_count),
         garbage_blob_bytes_(garbage_blob_bytes) {
     assert(shared_meta_);
@@ -145,6 +154,7 @@ class BlobFileMetaData {
   }
 
   std::shared_ptr<SharedBlobFileMetaData> shared_meta_;
+  LinkedSsts linked_ssts_;
   uint64_t garbage_blob_count_;
   uint64_t garbage_blob_bytes_;
 };

db/version_builder.cc

@@ -90,7 +90,8 @@ class VersionBuilder::Rep {
    public:
     bool IsEmpty() const {
       return !shared_meta_ && !additional_garbage_count_ &&
-             !additional_garbage_bytes_;
+             !additional_garbage_bytes_ && newly_linked_ssts_.empty() &&
+             newly_unlinked_ssts_.empty();
     }
 
     std::shared_ptr<SharedBlobFileMetaData> GetSharedMeta() const {
@@ -105,6 +106,14 @@ class VersionBuilder::Rep {
       return additional_garbage_bytes_;
     }
 
+    const std::unordered_set<uint64_t>& GetNewlyLinkedSsts() const {
+      return newly_linked_ssts_;
+    }
+
+    const std::unordered_set<uint64_t>& GetNewlyUnlinkedSsts() const {
+      return newly_unlinked_ssts_;
+    }
+
     void SetSharedMeta(std::shared_ptr<SharedBlobFileMetaData> shared_meta) {
       assert(!shared_meta_);
       assert(shared_meta);
@@ -117,10 +126,44 @@ class VersionBuilder::Rep {
       additional_garbage_bytes_ += bytes;
     }
 
+    void LinkSst(uint64_t sst_file_number) {
+      assert(newly_linked_ssts_.find(sst_file_number) ==
+             newly_linked_ssts_.end());
+
+      // Reconcile with newly unlinked SSTs on the fly. (Note: an SST can be
+      // linked to and unlinked from the same blob file in the case of a trivial
+      // move.)
+      auto it = newly_unlinked_ssts_.find(sst_file_number);
+
+      if (it != newly_unlinked_ssts_.end()) {
+        newly_unlinked_ssts_.erase(it);
+      } else {
+        newly_linked_ssts_.emplace(sst_file_number);
+      }
+    }
+
+    void UnlinkSst(uint64_t sst_file_number) {
+      assert(newly_unlinked_ssts_.find(sst_file_number) ==
+             newly_unlinked_ssts_.end());
+
+      // Reconcile with newly linked SSTs on the fly. (Note: an SST can be
+      // linked to and unlinked from the same blob file in the case of a trivial
+      // move.)
+      auto it = newly_linked_ssts_.find(sst_file_number);
+
+      if (it != newly_linked_ssts_.end()) {
+        newly_linked_ssts_.erase(it);
+      } else {
+        newly_unlinked_ssts_.emplace(sst_file_number);
+      }
+    }
+
    private:
     std::shared_ptr<SharedBlobFileMetaData> shared_meta_;
     uint64_t additional_garbage_count_ = 0;
     uint64_t additional_garbage_bytes_ = 0;
+    std::unordered_set<uint64_t> newly_linked_ssts_;
+    std::unordered_set<uint64_t> newly_unlinked_ssts_;
   };
 
   const FileOptions& file_options_;
@@ -212,29 +255,19 @@ class VersionBuilder::Rep {
     return false;
   }
 
-  Status CheckConsistencyOfOldestBlobFileReference(
-      const VersionStorageInfo* vstorage, uint64_t blob_file_number) const {
-    assert(vstorage);
+  using ExpectedLinkedSsts =
+      std::unordered_map<uint64_t, BlobFileMetaData::LinkedSsts>;
 
-    // TODO: remove this check once we actually start recoding metadata for
-    // blob files in the MANIFEST.
-    if (vstorage->GetBlobFiles().empty()) {
-      return Status::OK();
-    }
+  static void UpdateExpectedLinkedSsts(
+      uint64_t table_file_number, uint64_t blob_file_number,
+      ExpectedLinkedSsts* expected_linked_ssts) {
+    assert(expected_linked_ssts);
 
     if (blob_file_number == kInvalidBlobFileNumber) {
-      return Status::OK();
+      return;
     }
 
-    if (!IsBlobFileInVersion(blob_file_number)) {
-      std::ostringstream oss;
-      oss << "Blob file #" << blob_file_number
-          << " is not part of this version";
-
-      return Status::Corruption("VersionBuilder", oss.str());
-    }
-
-    return Status::OK();
+    (*expected_linked_ssts)[blob_file_number].emplace(table_file_number);
   }
 
   Status CheckConsistency(VersionStorageInfo* vstorage) {
@@ -245,9 +278,13 @@ class VersionBuilder::Rep {
       return Status::OK();
     }
 #endif
-    // Make sure the files are sorted correctly and that the oldest blob file
-    // reference for each table file points to a valid blob file in this
-    // version.
+    // Make sure the files are sorted correctly and that the links between
+    // table files and blob files are consistent. The latter is checked using
+    // the following mapping, which is built using the forward links
+    // (table file -> blob file), and is subsequently compared with the inverse
+    // mapping stored in the BlobFileMetaData objects.
+    ExpectedLinkedSsts expected_linked_ssts;
+
     for (int level = 0; level < num_levels_; level++) {
       auto& level_files = vstorage->LevelFiles(level);
 
@@ -256,19 +293,14 @@ class VersionBuilder::Rep {
       }
 
       assert(level_files[0]);
-      Status s = CheckConsistencyOfOldestBlobFileReference(
-          vstorage, level_files[0]->oldest_blob_file_number);
-      if (!s.ok()) {
-        return s;
-      }
-
+      UpdateExpectedLinkedSsts(level_files[0]->fd.GetNumber(),
+                               level_files[0]->oldest_blob_file_number,
+                               &expected_linked_ssts);
       for (size_t i = 1; i < level_files.size(); i++) {
         assert(level_files[i]);
-        s = CheckConsistencyOfOldestBlobFileReference(
-            vstorage, level_files[i]->oldest_blob_file_number);
-        if (!s.ok()) {
-          return s;
-        }
+        UpdateExpectedLinkedSsts(level_files[i]->fd.GetNumber(),
+                                 level_files[i]->oldest_blob_file_number,
+                                 &expected_linked_ssts);
 
         auto f1 = level_files[i - 1];
         auto f2 = level_files[i];
@@ -328,17 +360,27 @@ class VersionBuilder::Rep {
     // Make sure that all blob files in the version have non-garbage data.
     const auto& blob_files = vstorage->GetBlobFiles();
     for (const auto& pair : blob_files) {
+      const uint64_t blob_file_number = pair.first;
       const auto& blob_file_meta = pair.second;
       assert(blob_file_meta);
 
       if (blob_file_meta->GetGarbageBlobCount() >=
           blob_file_meta->GetTotalBlobCount()) {
         std::ostringstream oss;
-        oss << "Blob file #" << blob_file_meta->GetBlobFileNumber()
+        oss << "Blob file #" << blob_file_number
             << " consists entirely of garbage";
 
         return Status::Corruption("VersionBuilder", oss.str());
       }
+
+      if (blob_file_meta->GetLinkedSsts() !=
+          expected_linked_ssts[blob_file_number]) {
+        std::ostringstream oss;
+        oss << "Links are inconsistent between table files and blob file #"
+            << blob_file_number;
+
+        return Status::Corruption("VersionBuilder", oss.str());
+      }
     }
 
     Status ret_s;
@@ -430,6 +472,28 @@ class VersionBuilder::Rep {
     return base_vstorage_->GetFileLocation(file_number).GetLevel();
   }
 
+  uint64_t GetOldestBlobFileNumberForTableFile(int level,
+                                               uint64_t file_number) const {
+    assert(level < num_levels_);
+
+    const auto& added_files = levels_[level].added_files;
+
+    auto it = added_files.find(file_number);
+    if (it != added_files.end()) {
+      const FileMetaData* const meta = it->second;
+      assert(meta);
+
+      return meta->oldest_blob_file_number;
+    }
+
+    assert(base_vstorage_);
+    const FileMetaData* const meta =
+        base_vstorage_->GetFileMetaDataByNumber(file_number);
+    assert(meta);
+
+    return meta->oldest_blob_file_number;
+  }
+
   Status ApplyFileDeletion(int level, uint64_t file_number) {
     assert(level != VersionStorageInfo::FileLocation::Invalid().GetLevel());
 
@@ -463,6 +527,14 @@ class VersionBuilder::Rep {
       return Status::OK();
     }
 
+    const uint64_t blob_file_number =
+        GetOldestBlobFileNumberForTableFile(level, file_number);
+
+    if (blob_file_number != kInvalidBlobFileNumber &&
+        IsBlobFileInVersion(blob_file_number)) {
+      blob_file_meta_deltas_[blob_file_number].UnlinkSst(file_number);
+    }
+
     auto& level_state = levels_[level];
 
     auto& add_files = level_state.added_files;
@@ -523,6 +595,13 @@ class VersionBuilder::Rep {
     assert(add_files.find(file_number) == add_files.end());
     add_files.emplace(file_number, f);
 
+    const uint64_t blob_file_number = f->oldest_blob_file_number;
+
+    if (blob_file_number != kInvalidBlobFileNumber &&
+        IsBlobFileInVersion(blob_file_number)) {
+      blob_file_meta_deltas_[blob_file_number].LinkSst(file_number);
+    }
+
     table_file_levels_[file_number] = level;
 
     return Status::OK();
@@ -537,7 +616,27 @@ class VersionBuilder::Rep {
       }
     }
 
-    // Delete files
+    // Note: we process the blob file related changes first because the
+    // table file addition/deletion logic depends on the blob files
+    // already being there.
+
+    // Add new blob files
+    for (const auto& blob_file_addition : edit->GetBlobFileAdditions()) {
+      const Status s = ApplyBlobFileAddition(blob_file_addition);
+      if (!s.ok()) {
+        return s;
+      }
+    }
+
+    // Increase the amount of garbage for blob files affected by GC
+    for (const auto& blob_file_garbage : edit->GetBlobFileGarbages()) {
+      const Status s = ApplyBlobFileGarbage(blob_file_garbage);
+      if (!s.ok()) {
+        return s;
+      }
+    }
+
+    // Delete table files
     for (const auto& deleted_file : edit->GetDeletedFiles()) {
       const int level = deleted_file.first;
       const uint64_t file_number = deleted_file.second;
@@ -548,7 +647,7 @@ class VersionBuilder::Rep {
       }
     }
 
-    // Add new files
+    // Add new table files
     for (const auto& new_file : edit->GetNewFiles()) {
       const int level = new_file.first;
       const FileMetaData& meta = new_file.second;
@@ -559,23 +658,28 @@ class VersionBuilder::Rep {
       }
     }
 
-    // Add new blob files
-    for (const auto& blob_file_addition : edit->GetBlobFileAdditions()) {
-      const Status s = ApplyBlobFileAddition(blob_file_addition);
-      if (!s.ok()) {
-        return s;
-      }
+    return Status::OK();
+  }
+
+  static BlobFileMetaData::LinkedSsts ApplyLinkedSstChanges(
+      const BlobFileMetaData::LinkedSsts& base,
+      const std::unordered_set<uint64_t>& newly_linked,
+      const std::unordered_set<uint64_t>& newly_unlinked) {
+    BlobFileMetaData::LinkedSsts result(base);
+
+    for (uint64_t sst_file_number : newly_unlinked) {
+      assert(result.find(sst_file_number) != result.end());
+
+      result.erase(sst_file_number);
     }
 
-    // Increase the amount of garbage for blob files affected by GC
-    for (const auto& blob_file_garbage : edit->GetBlobFileGarbages()) {
-      const Status s = ApplyBlobFileGarbage(blob_file_garbage);
-      if (!s.ok()) {
-        return s;
-      }
+    for (uint64_t sst_file_number : newly_linked) {
+      assert(result.find(sst_file_number) == result.end());
+
+      result.emplace(sst_file_number);
     }
 
-    return Status::OK();
+    return result;
   }
 
   static std::shared_ptr<BlobFileMetaData> CreateMetaDataForNewBlobFile(
@@ -583,9 +687,11 @@ class VersionBuilder::Rep {
     auto shared_meta = delta.GetSharedMeta();
     assert(shared_meta);
 
-    auto meta = BlobFileMetaData::Create(std::move(shared_meta),
-                                         delta.GetAdditionalGarbageCount(),
-                                         delta.GetAdditionalGarbageBytes());
+    assert(delta.GetNewlyUnlinkedSsts().empty());
+
+    auto meta = BlobFileMetaData::Create(
+        std::move(shared_meta), delta.GetNewlyLinkedSsts(),
+        delta.GetAdditionalGarbageCount(), delta.GetAdditionalGarbageBytes());
 
     return meta;
   }
@@ -604,8 +710,12 @@ class VersionBuilder::Rep {
     auto shared_meta = base_meta->GetSharedMeta();
     assert(shared_meta);
 
+    auto linked_ssts = ApplyLinkedSstChanges(base_meta->GetLinkedSsts(),
+                                             delta.GetNewlyLinkedSsts(),
+                                             delta.GetNewlyUnlinkedSsts());
+
     auto meta = BlobFileMetaData::Create(
-        std::move(shared_meta),
+        std::move(shared_meta), std::move(linked_ssts),
         base_meta->GetGarbageBlobCount() + delta.GetAdditionalGarbageCount(),
         base_meta->GetGarbageBlobBytes() + delta.GetAdditionalGarbageBytes());
 
@@ -618,7 +728,8 @@ class VersionBuilder::Rep {
     assert(vstorage);
     assert(meta);
 
-    if (meta->GetGarbageBlobCount() < meta->GetTotalBlobCount()) {
+    if (meta->GetGarbageBlobCount() < meta->GetTotalBlobCount() ||
+        !meta->GetLinkedSsts().empty()) {
       vstorage->AddBlobFile(meta);
     }
   }

db/version_builder_test.cc

@@ -4,8 +4,11 @@
 //  (found in the LICENSE.Apache file in the root directory).
 
 #include <cstring>
+#include <iomanip>
 #include <memory>
+#include <sstream>
 #include <string>
+
 #include "db/version_edit.h"
 #include "db/version_set.h"
 #include "logging/logging.h"
@@ -80,13 +83,15 @@ class VersionBuilderTest : public testing::Test {
 
   void AddBlob(uint64_t blob_file_number, uint64_t total_blob_count,
                uint64_t total_blob_bytes, std::string checksum_method,
-               std::string checksum_value, uint64_t garbage_blob_count,
-               uint64_t garbage_blob_bytes) {
+               std::string checksum_value,
+               BlobFileMetaData::LinkedSsts linked_ssts,
+               uint64_t garbage_blob_count, uint64_t garbage_blob_bytes) {
     auto shared_meta = SharedBlobFileMetaData::Create(
         blob_file_number, total_blob_count, total_blob_bytes,
         std::move(checksum_method), std::move(checksum_value));
-    auto meta = BlobFileMetaData::Create(
-        std::move(shared_meta), garbage_blob_count, garbage_blob_bytes);
+    auto meta =
+        BlobFileMetaData::Create(std::move(shared_meta), std::move(linked_ssts),
+                                 garbage_blob_count, garbage_blob_bytes);
 
     vstorage_.AddBlobFile(std::move(meta));
   }
@@ -642,6 +647,7 @@ 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->GetGarbageBlobCount(), 0);
   ASSERT_EQ(new_meta->GetGarbageBlobBytes(), 0);
 }
@@ -658,7 +664,8 @@ TEST_F(VersionBuilderTest, ApplyBlobFileAdditionAlreadyInBase) {
   constexpr uint64_t garbage_blob_bytes = 456789;
 
   AddBlob(blob_file_number, total_blob_count, total_blob_bytes, checksum_method,
-          checksum_value, garbage_blob_count, garbage_blob_bytes);
+          checksum_value, BlobFileMetaData::LinkedSsts(), garbage_blob_count,
+          garbage_blob_bytes);
 
   EnvOptions env_options;
   constexpr TableCache* table_cache = nullptr;
@@ -717,7 +724,8 @@ 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, garbage_blob_count, garbage_blob_bytes);
+          checksum_value, BlobFileMetaData::LinkedSsts(), garbage_blob_count,
+          garbage_blob_bytes);
 
   const auto meta =
       GetBlobFileMetaData(vstorage_.GetBlobFiles(), blob_file_number);
@@ -759,6 +767,7 @@ 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->GetGarbageBlobCount(),
             garbage_blob_count + new_garbage_blob_count);
   ASSERT_EQ(new_meta->GetGarbageBlobBytes(),
@@ -816,6 +825,7 @@ 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->GetGarbageBlobCount(), garbage_blob_count);
   ASSERT_EQ(new_meta->GetGarbageBlobBytes(), garbage_blob_bytes);
 }
@@ -856,8 +866,8 @@ TEST_F(VersionBuilderTest, SaveBlobFilesTo) {
 
     AddBlob(blob_file_number, total_blob_count, total_blob_bytes,
             /* checksum_method */ std::string(),
-            /* checksum_value */ std::string(), garbage_blob_count,
-            garbage_blob_bytes);
+            /* checksum_value */ std::string(), BlobFileMetaData::LinkedSsts(),
+            garbage_blob_count, garbage_blob_bytes);
   }
 
   EnvOptions env_options;
@@ -946,7 +956,7 @@ TEST_F(VersionBuilderTest, CheckConsistencyForBlobFiles) {
   AddBlob(/* blob_file_number */ 16, /* total_blob_count */ 1000,
           /* total_blob_bytes */ 1000000,
           /* checksum_method */ std::string(),
-          /* checksum_value */ std::string(),
+          /* checksum_value */ std::string(), BlobFileMetaData::LinkedSsts{1},
           /* garbage_blob_count */ 500, /* garbage_blob_bytes */ 300000);
 
   UpdateVersionStorageInfo();
@@ -995,9 +1005,9 @@ TEST_F(VersionBuilderTest, CheckConsistencyForBlobFiles) {
   UnrefFilesInVersion(&new_vstorage);
 }
 
-TEST_F(VersionBuilderTest, CheckConsistencyForBlobFilesNotInVersion) {
-  // Initialize base version. The table file points to a blob file that is
-  // not in this version.
+TEST_F(VersionBuilderTest, CheckConsistencyForBlobFilesInconsistentLinks) {
+  // Initialize base version. Links between the table file and the blob file
+  // are inconsistent.
 
   Add(/* level */ 1, /* file_number */ 1, /* smallest */ "150",
       /* largest */ "200", /* file_size */ 100,
@@ -1009,7 +1019,7 @@ TEST_F(VersionBuilderTest, CheckConsistencyForBlobFilesNotInVersion) {
   AddBlob(/* blob_file_number */ 16, /* total_blob_count */ 1000,
           /* total_blob_bytes */ 1000000,
           /* checksum_method */ std::string(),
-          /* checksum_value */ std::string(),
+          /* checksum_value */ std::string(), BlobFileMetaData::LinkedSsts{1},
           /* garbage_blob_count */ 500, /* garbage_blob_bytes */ 300000);
 
   UpdateVersionStorageInfo();
@@ -1029,8 +1039,9 @@ TEST_F(VersionBuilderTest, CheckConsistencyForBlobFilesNotInVersion) {
 
   const Status s = builder.SaveTo(&new_vstorage);
   ASSERT_TRUE(s.IsCorruption());
-  ASSERT_TRUE(
-      std::strstr(s.getState(), "Blob file #256 is not part of this version"));
+  ASSERT_TRUE(std::strstr(
+      s.getState(),
+      "Links are inconsistent between table files and blob file #16"));
 
   UnrefFilesInVersion(&new_vstorage);
 }
@@ -1049,7 +1060,7 @@ TEST_F(VersionBuilderTest, CheckConsistencyForBlobFilesAllGarbage) {
   AddBlob(/* blob_file_number */ 16, /* total_blob_count */ 1000,
           /* total_blob_bytes */ 1000000,
           /* checksum_method */ std::string(),
-          /* checksum_value */ std::string(),
+          /* checksum_value */ std::string(), BlobFileMetaData::LinkedSsts{1},
           /* garbage_blob_count */ 1000, /* garbage_blob_bytes */ 1000000);
 
   UpdateVersionStorageInfo();
@@ -1075,6 +1086,224 @@ TEST_F(VersionBuilderTest, CheckConsistencyForBlobFilesAllGarbage) {
   UnrefFilesInVersion(&new_vstorage);
 }
 
+TEST_F(VersionBuilderTest, CheckConsistencyForBlobFilesAllGarbageLinkedSsts) {
+  // Initialize base version, with a table file pointing to a blob file
+  // that has no garbage at this point.
+
+  Add(/* level */ 1, /* file_number */ 1, /* smallest */ "150",
+      /* largest */ "200", /* file_size */ 100,
+      /* path_id */ 0, /* smallest_seq */ 100, /* largest_seq */ 100,
+      /* num_entries */ 0, /* num_deletions */ 0,
+      /* sampled */ false, /* smallest_seqno */ 100, /* largest_seqno */ 100,
+      /* oldest_blob_file_number */ 16);
+
+  AddBlob(/* blob_file_number */ 16, /* total_blob_count */ 1000,
+          /* total_blob_bytes */ 1000000,
+          /* checksum_method */ std::string(),
+          /* checksum_value */ std::string(), BlobFileMetaData::LinkedSsts{1},
+          /* garbage_blob_count */ 0, /* garbage_blob_bytes */ 0);
+
+  UpdateVersionStorageInfo();
+
+  // Mark the entire blob file garbage but do not remove the linked SST.
+  EnvOptions env_options;
+  constexpr TableCache* table_cache = nullptr;
+  constexpr VersionSet* version_set = nullptr;
+
+  VersionBuilder builder(env_options, &ioptions_, table_cache, &vstorage_,
+                         version_set);
+
+  VersionEdit edit;
+
+  edit.AddBlobFileGarbage(/* blob_file_number */ 16,
+                          /* garbage_blob_count */ 1000,
+                          /* garbage_blob_bytes */ 1000000);
+
+  ASSERT_OK(builder.Apply(&edit));
+
+  // Save to a new version in order to trigger consistency checks.
+  constexpr bool force_consistency_checks = true;
+  VersionStorageInfo new_vstorage(&icmp_, ucmp_, options_.num_levels,
+                                  kCompactionStyleLevel, &vstorage_,
+                                  force_consistency_checks);
+
+  const Status s = builder.SaveTo(&new_vstorage);
+  ASSERT_TRUE(s.IsCorruption());
+  ASSERT_TRUE(
+      std::strstr(s.getState(), "Blob file #16 consists entirely of garbage"));
+
+  UnrefFilesInVersion(&new_vstorage);
+}
+
+TEST_F(VersionBuilderTest, MaintainLinkedSstsForBlobFiles) {
+  // Initialize base version. Table files 1..10 are linked to blob files 1..5,
+  // while table files 11..20 are not linked to any blob files.
+
+  for (uint64_t i = 1; i <= 10; ++i) {
+    std::ostringstream oss;
+    oss << std::setw(2) << std::setfill('0') << i;
+
+    const std::string key = oss.str();
+
+    Add(/* level */ 1, /* file_number */ i, /* smallest */ key.c_str(),
+        /* largest */ key.c_str(), /* file_size */ 100,
+        /* path_id */ 0, /* smallest_seq */ i * 100, /* largest_seq */ i * 100,
+        /* num_entries */ 0, /* num_deletions */ 0,
+        /* sampled */ false, /* smallest_seqno */ i * 100,
+        /* largest_seqno */ i * 100,
+        /* oldest_blob_file_number */ ((i - 1) % 5) + 1);
+  }
+
+  for (uint64_t i = 1; i <= 5; ++i) {
+    AddBlob(/* blob_file_number */ i, /* total_blob_count */ 2000,
+            /* total_blob_bytes */ 2000000,
+            /* checksum_method */ std::string(),
+            /* checksum_value */ std::string(),
+            BlobFileMetaData::LinkedSsts{i, i + 5},
+            /* garbage_blob_count */ 1000, /* garbage_blob_bytes */ 1000000);
+  }
+
+  for (uint64_t i = 11; i <= 20; ++i) {
+    std::ostringstream oss;
+    oss << std::setw(2) << std::setfill('0') << i;
+
+    const std::string key = oss.str();
+
+    Add(/* level */ 1, /* file_number */ i, /* smallest */ key.c_str(),
+        /* largest */ key.c_str(), /* file_size */ 100,
+        /* path_id */ 0, /* smallest_seq */ i * 100, /* largest_seq */ i * 100,
+        /* num_entries */ 0, /* num_deletions */ 0,
+        /* sampled */ false, /* smallest_seqno */ i * 100,
+        /* largest_seqno */ i * 100, kInvalidBlobFileNumber);
+  }
+
+  UpdateVersionStorageInfo();
+
+  {
+    const auto& blob_files = vstorage_.GetBlobFiles();
+    ASSERT_EQ(blob_files.size(), 5);
+
+    const std::vector<BlobFileMetaData::LinkedSsts> expected_linked_ssts{
+        {1, 6}, {2, 7}, {3, 8}, {4, 9}, {5, 10}};
+
+    for (size_t i = 0; i < 5; ++i) {
+      const auto meta =
+          GetBlobFileMetaData(blob_files, /* blob_file_number */ i + 1);
+      ASSERT_NE(meta, nullptr);
+      ASSERT_EQ(meta->GetLinkedSsts(), expected_linked_ssts[i]);
+    }
+  }
+
+  VersionEdit edit;
+
+  // Add an SST that references a blob file.
+  edit.AddFile(
+      /* level */ 1, /* file_number */ 21, /* path_id */ 0,
+      /* file_size */ 100, /* smallest */ GetInternalKey("21", 2100),
+      /* largest */ GetInternalKey("21", 2100), /* smallest_seqno */ 2100,
+      /* largest_seqno */ 2100, /* marked_for_compaction */ false,
+      /* oldest_blob_file_number */ 1, kUnknownOldestAncesterTime,
+      kUnknownFileCreationTime, kUnknownFileChecksum,
+      kUnknownFileChecksumFuncName);
+
+  // Add an SST that does not reference any blob files.
+  edit.AddFile(
+      /* level */ 1, /* file_number */ 22, /* path_id */ 0,
+      /* file_size */ 100, /* smallest */ GetInternalKey("22", 2200),
+      /* largest */ GetInternalKey("22", 2200), /* smallest_seqno */ 2200,
+      /* largest_seqno */ 2200, /* marked_for_compaction */ false,
+      kInvalidBlobFileNumber, kUnknownOldestAncesterTime,
+      kUnknownFileCreationTime, kUnknownFileChecksum,
+      kUnknownFileChecksumFuncName);
+
+  // Delete a file that references a blob file.
+  edit.DeleteFile(/* level */ 1, /* file_number */ 6);
+
+  // Delete a file that does not reference any blob files.
+  edit.DeleteFile(/* level */ 1, /* file_number */ 16);
+
+  // Trivially move a file that references a blob file. Note that we save
+  // the original BlobFileMetaData object so we can check that no new object
+  // gets created.
+  auto meta3 =
+      GetBlobFileMetaData(vstorage_.GetBlobFiles(), /* blob_file_number */ 3);
+
+  edit.DeleteFile(/* level */ 1, /* file_number */ 3);
+  edit.AddFile(/* level */ 2, /* file_number */ 3, /* path_id */ 0,
+               /* file_size */ 100, /* smallest */ GetInternalKey("03", 300),
+               /* largest */ GetInternalKey("03", 300),
+               /* smallest_seqno */ 300,
+               /* largest_seqno */ 300, /* marked_for_compaction */ false,
+               /* oldest_blob_file_number */ 3, kUnknownOldestAncesterTime,
+               kUnknownFileCreationTime, kUnknownFileChecksum,
+               kUnknownFileChecksumFuncName);
+
+  // Trivially move a file that does not reference any blob files.
+  edit.DeleteFile(/* level */ 1, /* file_number */ 13);
+  edit.AddFile(/* level */ 2, /* file_number */ 13, /* path_id */ 0,
+               /* file_size */ 100, /* smallest */ GetInternalKey("13", 1300),
+               /* largest */ GetInternalKey("13", 1300),
+               /* smallest_seqno */ 1300,
+               /* largest_seqno */ 1300, /* marked_for_compaction */ false,
+               kInvalidBlobFileNumber, kUnknownOldestAncesterTime,
+               kUnknownFileCreationTime, kUnknownFileChecksum,
+               kUnknownFileChecksumFuncName);
+
+  // Add one more SST file that references a blob file, then promptly
+  // delete it in a second version edit before the new version gets saved.
+  // This file should not show up as linked to the blob file in the new version.
+  edit.AddFile(/* level */ 1, /* file_number */ 23, /* path_id */ 0,
+               /* file_size */ 100, /* smallest */ GetInternalKey("23", 2300),
+               /* largest */ GetInternalKey("23", 2300),
+               /* smallest_seqno */ 2300,
+               /* largest_seqno */ 2300, /* marked_for_compaction */ false,
+               /* oldest_blob_file_number */ 5, kUnknownOldestAncesterTime,
+               kUnknownFileCreationTime, kUnknownFileChecksum,
+               kUnknownFileChecksumFuncName);
+
+  VersionEdit edit2;
+
+  edit2.DeleteFile(/* level */ 1, /* file_number */ 23);
+
+  EnvOptions env_options;
+  constexpr TableCache* table_cache = nullptr;
+  constexpr VersionSet* version_set = nullptr;
+
+  VersionBuilder builder(env_options, &ioptions_, table_cache, &vstorage_,
+                         version_set);
+
+  ASSERT_OK(builder.Apply(&edit));
+  ASSERT_OK(builder.Apply(&edit2));
+
+  constexpr bool force_consistency_checks = true;
+  VersionStorageInfo new_vstorage(&icmp_, ucmp_, options_.num_levels,
+                                  kCompactionStyleLevel, &vstorage_,
+                                  force_consistency_checks);
+
+  ASSERT_OK(builder.SaveTo(&new_vstorage));
+
+  {
+    const auto& blob_files = new_vstorage.GetBlobFiles();
+    ASSERT_EQ(blob_files.size(), 5);
+
+    const std::vector<BlobFileMetaData::LinkedSsts> expected_linked_ssts{
+        {1, 21}, {2, 7}, {3, 8}, {4, 9}, {5, 10}};
+
+    for (size_t i = 0; i < 5; ++i) {
+      const auto meta =
+          GetBlobFileMetaData(blob_files, /* blob_file_number */ i + 1);
+      ASSERT_NE(meta, nullptr);
+      ASSERT_EQ(meta->GetLinkedSsts(), expected_linked_ssts[i]);
+    }
+
+    // Make sure that no new BlobFileMetaData got created for the blob file
+    // affected by the trivial move.
+    ASSERT_EQ(GetBlobFileMetaData(blob_files, /* blob_file_number */ 3), meta3);
+  }
+
+  UnrefFilesInVersion(&new_vstorage);
+}
+
 TEST_F(VersionBuilderTest, CheckConsistencyForFileDeletedTwice) {
   Add(0, 1U, "150", "200", 100U);
   UpdateVersionStorageInfo();