@@ -1897,6 +1897,66 @@ TEST_F(DBBlobBasicTest, BestEffortsBlobDbRepairWithAtomicFlush) {
   ASSERT_EQ(Get(1, "y"), "vy");
 }
 
+TEST_F(DBBlobBasicTest,
+       BestEffortsBlobDbRepairAutomaticCompactionDiscardsDroppedValues) {
+  Options options = MissingBlobRepairOptions();
+  uint64_t missing_blob_file = 0;
+  ASSERT_NO_FATAL_FAILURE(
+      RepairMissingNonOldestBlob(options, &missing_blob_file));
+  ASSERT_EQ(NumTableFilesAtLevel(1), 1);
+
+  // Reopen without repair. Overwriting "b" makes an automatic compaction drop
+  // its old value, which is in the dropped blob file.
+  options.disable_auto_compactions = false;
+  options.level0_file_num_compaction_trigger = 1;
+  Reopen(options);
+  ASSERT_OK(Put("b", "vb2"));
+  ASSERT_OK(Flush());
+  ASSERT_OK(dbfull()->TEST_WaitForCompact());
+  ASSERT_EQ(NumTableFilesAtLevel(0), 0);
+  uint64_t background_errors = 0;
+  ASSERT_TRUE(db_->GetIntProperty(DB::Properties::kBackgroundErrors,
+                                  &background_errors));
+  ASSERT_EQ(background_errors, 0);
+
+  ASSERT_EQ(Get("a"), "va");
+  ASSERT_EQ(Get("b"), "vb2");
+  ASSERT_EQ(Get("c"), "vc");
+  ASSERT_EQ(Get("d"), "NOT_FOUND");
+
+  Reopen(options);
+  ASSERT_EQ(Get("b"), "vb2");
+  ASSERT_EQ(Get("d"), "NOT_FOUND");
+}
+
+TEST_F(DBBlobBasicTest, BestEffortsBlobDbRepairWithBlobGarbageCollection) {
+  Options options = MissingBlobRepairOptions();
+  uint64_t missing_blob_file = 0;
+  ASSERT_NO_FATAL_FAILURE(
+      RepairMissingNonOldestBlob(options, &missing_blob_file));
+
+  // Garbage collection relocates every blob, so it would read the dropped
+  // blob file if the compaction did not discard its values first.
+  options.enable_blob_garbage_collection = true;
+  options.blob_garbage_collection_age_cutoff = 1.0;
+  Reopen(options);
+  CompactRangeOptions compact_range_options;
+  compact_range_options.bottommost_level_compaction =
+      BottommostLevelCompaction::kForce;
+  ASSERT_OK(db_->CompactRange(compact_range_options, nullptr, nullptr));
+  ASSERT_EQ(Get("a"), "va");
+  ASSERT_EQ(Get("c"), "vc");
+  ASSERT_EQ(Get("b"), "NOT_FOUND");
+  ASSERT_EQ(Get("d"), "NOT_FOUND");
+
+  // The surviving values were relocated to a new blob file, so no SST file
+  // can reference the dropped blob file anymore.
+  ASSERT_TRUE(GetDroppedBlobFiles().empty());
+  Reopen(options);
+  ASSERT_TRUE(GetDroppedBlobFiles().empty());
+  ASSERT_EQ(Get("a"), "va");
+}
+
 TEST_F(DBBlobBasicTest, BestEffortsBlobDbRepairMissingOldestBlobWithForcedGc) {
   // After repair drops the oldest blob file an SST file references, the
   // oldest blob file left in the Version has no linked SST file.
 
@@ -20,16 +20,20 @@
 namespace ROCKSDB_NAMESPACE {
 
 // Replaces references absent from an immutable input Version with deletions
-// or entities containing the surviving columns. Input values must already be
+// or entities containing the surviving columns. With `only_dropped_blob_files`,
+// only references to blob files recorded as dropped in the input Version (see
+// VersionStorageInfo::GetDroppedBlobFiles()) are replaced; otherwise
+// references to any blob file absent from it are. Input values must already be
 // prepared, as they are for compaction input. Neither input is owned.
 class MissingBlobRepairIterator : public InternalIterator {
  public:
   MissingBlobRepairIterator(InternalIterator* input,
                             const VersionStorageInfo* input_storage,
-                            size_t timestamp_size)
+                            size_t timestamp_size, bool only_dropped_blob_files)
       : input_(input),
         input_storage_(input_storage),
-        timestamp_size_(timestamp_size) {
+        timestamp_size_(timestamp_size),
+        only_dropped_blob_files_(only_dropped_blob_files) {
     assert(input_);
     assert(input_storage_);
     Update();
@@ -161,6 +165,9 @@ class MissingBlobRepairIterator : public InternalIterator {
     if (index.IsInlined() || index.HasTTL() || index.IsSameFile()) {
       return false;
     }
+    if (only_dropped_blob_files_) {
+      return input_storage_->IsDroppedBlobFile(index.file_number());
+    }
     const VersionStorageInfo::BlobFiles::const_iterator it =
         input_storage_->GetBlobFileMetaDataLB(index.file_number());
     return it == input_storage_->GetBlobFiles().end() ||
@@ -182,6 +189,20 @@ class MissingBlobRepairIterator : public InternalIterator {
   // changes the positions of surviving blob indexes, so rebuild those positions
   // before serialization. Only loss of every column becomes a deletion.
   Status RepairEntity(const ParsedInternalKey& parsed) {
+    if (only_dropped_blob_files_) {
+      // Most entities reference no dropped blob file. Find out from the blob
+      // indexes alone before paying for a full deserialization.
+      bool has_dropped = false;
+      const Status s = WideColumnSerialization::ForEachBlobFileNumber(
+          input_->value(), [this, &has_dropped](const BlobIndex& index) {
+            has_dropped |= IsMissing(index);
+            return Status::OK();
+          });
+      if (!s.ok() || !has_dropped) {
+        return s;
+      }
+    }
+
     columns_.clear();
     blob_columns_.clear();
     Status s = WideColumnSerialization::Deserialize(input_->value(), columns_,
@@ -277,6 +298,7 @@ class MissingBlobRepairIterator : public InternalIterator {
   WideColumns surviving_columns_;
   std::vector<std::pair<size_t, BlobIndex>> surviving_blobs_;
   Status status_;
+  bool only_dropped_blob_files_;
   bool value_rewritten_ = false;
   bool repair_deletion_ = false;
 };
 
@@ -972,6 +972,38 @@ bool Compaction::DoesInputReferenceBlobFiles() const {
   return false;
 }
 
+bool Compaction::RepairsMissingBlobReferences() const {
+  if (skip_missing_blob_files_) {
+    return true;
+  }
+
+  assert(input_version_);
+  const VersionStorageInfo* storage_info = input_version_->storage_info();
+  assert(storage_info);
+
+  const std::vector<uint64_t>& dropped_blob_files =
+      storage_info->GetDroppedBlobFiles();
+  if (dropped_blob_files.empty()) {
+    return false;
+  }
+
+  // An SST file only references blob files at or above its oldest referenced
+  // blob file.
+  const uint64_t newest_dropped_blob_file = dropped_blob_files.back();
+  for (const CompactionInputFiles& level_inputs : inputs_) {
+    for (const FileMetaData* meta : level_inputs.files) {
+      assert(meta);
+
+      if (meta->oldest_blob_file_number != kInvalidBlobFileNumber &&
+          meta->oldest_blob_file_number <= newest_dropped_blob_file) {
+        return true;
+      }
+    }
+  }
+
+  return false;
+}
+
 uint64_t Compaction::MaxInputFileNewestKeyTime(const InternalKey* start,
                                                const InternalKey* end) const {
   uint64_t newest_key_time = kUnknownNewestKeyTime;
 
@@ -362,6 +362,16 @@ class Compaction {
   // PRE: input version has been set.
   bool DoesInputReferenceBlobFiles() const;
 
+  // Returns true iff this compaction replaces references to blob files missing
+  // from its input Version (see MissingBlobRepairIterator): with
+  // skip_missing_blob_files(), references to any such blob file; otherwise
+  // references to blob files recorded as dropped in the input Version (see
+  // VersionStorageInfo::GetDroppedBlobFiles()), when an input file might
+  // reference one.
+  //
+  // PRE: input version has been set.
+  bool RepairsMissingBlobReferences() const;
+
   // test function to validate the functionality of IsBottommostLevel()
   // function -- determines if compaction with inputs and storage is bottommost
   static bool TEST_IsBottommostLevel(
 
@@ -321,7 +321,8 @@ class CompactionIteratorTest : public testing::TestWithParam<bool> {
     InternalIterator* input = iter_.get();
     if (repair_enabled_) {
       repair_iter_ = std::make_unique<MissingBlobRepairIterator>(
-          input, repair_storage_.get(), cmp_->timestamp_size());
+          input, repair_storage_.get(), cmp_->timestamp_size(),
+          repair_only_dropped_blob_files_);
       input = repair_iter_.get();
     }
     c_iter_.reset(new CompactionIterator(
@@ -338,6 +339,15 @@ class CompactionIteratorTest : public testing::TestWithParam<bool> {
     c_iter_->SetMissingBlobRepairIterator(repair_iter_.get());
   }
 
+  // Repairs only references to `dropped_files`, which are recorded as dropped
+  // in a Version that holds `healthy_files`.
+  void EnableDroppedBlobRepair(const std::vector<uint64_t>& healthy_files,
+                               std::vector<uint64_t> dropped_files) {
+    EnableMissingBlobRepair(healthy_files);
+    repair_storage_->SetDroppedBlobFiles(std::move(dropped_files));
+    repair_only_dropped_blob_files_ = true;
+  }
+
   void EnableMissingBlobRepair(const std::vector<uint64_t>& healthy_files) {
     repair_enabled_ = true;
     repair_storage_ = std::make_unique<VersionStorageInfo>(
@@ -415,6 +425,7 @@ class CompactionIteratorTest : public testing::TestWithParam<bool> {
   const std::atomic<bool> kManualCompactionCanceledFalse_{false};
   FakeCompaction* compaction_proxy_;
   bool repair_enabled_ = false;
+  bool repair_only_dropped_blob_files_ = false;
 };
 
 TEST_P(CompactionIteratorTest, MergeBeforeFilterFullOrder) {
@@ -1088,7 +1099,8 @@ TEST_P(CompactionIteratorTest, MissingBlobRepairPreservesInputErrors) {
   std::unique_ptr<InternalIterator> failed_input(
       NewErrorInternalIterator(Status::IOError("Failed SST read")));
   MissingBlobRepairIterator failed_repair(failed_input.get(),
-                                          repair_storage_.get(), 0);
+                                          repair_storage_.get(), 0,
+                                          /*only_dropped_blob_files=*/false);
   failed_repair.SeekToFirst();
   ASSERT_FALSE(failed_repair.Valid());
   ASSERT_EQ(failed_repair.status().code(), Status::Code::kIOError);
@@ -1129,7 +1141,8 @@ TEST_P(CompactionIteratorTest, MissingBlobRepairPreservesInputErrors) {
 
   FailingNextIterator input({test::KeyStr("a", 5, kTypeBlobIndex)},
                             {BlobIndexForFile(20)}, &icmp_);
-  MissingBlobRepairIterator repair(&input, repair_storage_.get(), 0);
+  MissingBlobRepairIterator repair(&input, repair_storage_.get(), 0,
+                                   /*only_dropped_blob_files=*/false);
   repair.SeekToFirst();
   ASSERT_OK(repair.status());
   ASSERT_TRUE(repair.Valid());
@@ -1141,12 +1154,84 @@ TEST_P(CompactionIteratorTest, MissingBlobRepairPreservesInputErrors) {
   ASSERT_FALSE(repair.IsRepairDeletion());
 }
 
+TEST_P(CompactionIteratorTest, MissingBlobRepairOnlyDroppedBlobFiles) {
+  // Blob file 10 is in the Version, 20 is recorded as dropped and 30 is
+  // missing without a record.
+  EnableMissingBlobRepair({10});
+  repair_storage_->SetDroppedBlobFiles({20});
+  VectorIterator input(
+      {test::KeyStr("a", 5, kTypeBlobIndex),
+       test::KeyStr("b", 4, kTypeBlobIndex),
+       test::KeyStr("c", 3, kTypeBlobIndex)},
+      {BlobIndexForFile(10), BlobIndexForFile(20), BlobIndexForFile(30)},
+      &icmp_);
+  MissingBlobRepairIterator repair(&input, repair_storage_.get(), 0,
+                                   /*only_dropped_blob_files=*/true);
+  repair.SeekToFirst();
+  ASSERT_TRUE(repair.Valid());
+  ASSERT_EQ(repair.key(), test::KeyStr("a", 5, kTypeBlobIndex));
+  ASSERT_FALSE(repair.IsRepairDeletion());
+  repair.Next();
+  ASSERT_TRUE(repair.Valid());
+  ASSERT_EQ(repair.key(), test::KeyStr("b", 4, kTypeDeletion));
+  ASSERT_TRUE(repair.IsRepairDeletion());
+  repair.Next();
+  ASSERT_TRUE(repair.Valid());
+  ASSERT_EQ(repair.key(), test::KeyStr("c", 3, kTypeBlobIndex));
+  ASSERT_EQ(repair.value(), BlobIndexForFile(30));
+  ASSERT_FALSE(repair.IsRepairDeletion());
+  repair.Next();
+  ASSERT_FALSE(repair.Valid());
+  ASSERT_OK(repair.status());
+}
+
+TEST_P(CompactionIteratorTest, MissingBlobRepairOnlyDroppedWideColumns) {
+  // Blob file 20 is in the Version, 30 is recorded as dropped and 40 is
+  // missing without a record.
+  EnableDroppedBlobRepair({20}, {30});
+  BlobIndex healthy;
+  BlobIndex dropped;
+  BlobIndex unrecorded;
+  ASSERT_OK(healthy.DecodeFrom(BlobIndexForFile(20)));
+  ASSERT_OK(dropped.DecodeFrom(BlobIndexForFile(30)));
+  ASSERT_OK(unrecorded.DecodeFrom(BlobIndexForFile(40)));
+
+  // Only the column in the dropped blob file is removed.
+  std::string mixed;
+  ASSERT_OK(WideColumnSerialization::SerializeV2(
+      WideColumns{{"", "x"}, {"c1", ""}, {"c2", ""}, {"c3", ""}},
+      {{1, dropped}, {2, healthy}, {3, unrecorded}}, mixed));
+  std::string mixed_repaired;
+  ASSERT_OK(WideColumnSerialization::SerializeV2(
+      WideColumns{{"", "x"}, {"c2", ""}, {"c3", ""}},
+      {{1, healthy}, {2, unrecorded}}, mixed_repaired));
+  // An entity without dropped columns is left as is.
+  std::string healthy_only;
+  ASSERT_OK(WideColumnSerialization::SerializeV2(WideColumns{{"c2", ""}},
+                                                 {{0, healthy}}, healthy_only));
+  // An entity that loses every column becomes a deletion.
+  std::string dropped_only;
+  ASSERT_OK(WideColumnSerialization::SerializeV2(WideColumns{{"c1", ""}},
+                                                 {{0, dropped}}, dropped_only));
+
+  RunTest({test::KeyStr("a", 5, kTypeWideColumnEntity),
+           test::KeyStr("b", 5, kTypeWideColumnEntity),
+           test::KeyStr("c", 5, kTypeWideColumnEntity),
+           test::KeyStr("c", 3, kTypeValue)},
+          {mixed, healthy_only, dropped_only, "older"},
+          {test::KeyStr("a", 5, kTypeWideColumnEntity),
+           test::KeyStr("b", 5, kTypeWideColumnEntity),
+           test::KeyStr("c", 5, kTypeDeletion)},
+          {mixed_repaired, healthy_only, ""});
+}
+
 TEST_P(CompactionIteratorTest, MissingBlobRepairSeekAndNextUseRewrittenKeys) {
   EnableMissingBlobRepair({});
   VectorIterator input(
       {test::KeyStr("a", 5, kTypeBlobIndex), test::KeyStr("b", 4, kTypeValue)},
       {BlobIndexForFile(10), "healthy"}, &icmp_);
-  MissingBlobRepairIterator repair(&input, repair_storage_.get(), 0);
+  MissingBlobRepairIterator repair(&input, repair_storage_.get(), 0,
+                                   /*only_dropped_blob_files=*/false);
   const std::string deletion = test::KeyStr("a", 5, kTypeDeletion);
   repair.Seek(deletion);
   ASSERT_OK(repair.status());
@@ -1173,14 +1258,16 @@ TEST_P(CompactionIteratorTest,
   {
     std::unique_ptr<InternalIterator> input(
         NewErrorInternalIterator(Status::IOError("Failed SST read")));
-    MissingBlobRepairIterator repair(input.get(), repair_storage_.get(), 0);
+    MissingBlobRepairIterator repair(input.get(), repair_storage_.get(), 0,
+                                     /*only_dropped_blob_files=*/false);
     ASSERT_FALSE(repair.Valid());
   }
   VectorIterator input({test::KeyStr("a", 5, kTypeWideColumnEntity)},
                        {"invalid entity"});
   input.SeekToFirst();
   ASSERT_TRUE(input.Valid());
-  MissingBlobRepairIterator repair(&input, repair_storage_.get(), 0);
+  MissingBlobRepairIterator repair(&input, repair_storage_.get(), 0,
+                                   /*only_dropped_blob_files=*/false);
 }
 
 // It is possible that the output of the compaction iterator is empty even if
@@ -2321,7 +2408,8 @@ TEST_P(CompactionIteratorTsGcTest,
   input.SeekToFirst();
   ASSERT_TRUE(input.Valid());
   MissingBlobRepairIterator repair(&input, repair_storage_.get(),
-                                   cmp_->timestamp_size());
+                                   cmp_->timestamp_size(),
+                                   /*only_dropped_blob_files=*/false);
 }
 
 TEST_P(CompactionIteratorTsGcTest, MissingBlobRepairRespectsHistoryTrimming) {
@@ -2336,7 +2424,8 @@ TEST_P(CompactionIteratorTsGcTest, MissingBlobRepairRespectsHistoryTrimming) {
   PutFixed64(&trim_timestamp, 103);
   HistoryTrimmingIterator trimmed(&input, cmp_, trim_timestamp);
   MissingBlobRepairIterator repair(&trimmed, repair_storage_.get(),
-                                   cmp_->timestamp_size());
+                                   cmp_->timestamp_size(),
+                                   /*only_dropped_blob_files=*/false);
   repair.SeekToFirst();
   ASSERT_OK(repair.status());
   ASSERT_TRUE(repair.Valid());