Fix missing obsolete sequence for deletion in deleteblobfiles (#212) (#213)

src/blob_file_set.cc

@@ -294,7 +294,7 @@ Status BlobFileSet::DeleteBlobFilesInRanges(uint32_t cf_id,
     Status s = it->second->GetBlobFilesInRanges(ranges, n, include_end, &files);
     if (!s.ok()) return s;
     for (auto file_number : files) {
-      edit.DeleteBlobFile(file_number);
+      edit.DeleteBlobFile(file_number, obsolete_sequence);
     }
     s = LogAndApply(edit);
     return s;

src/titan_db_test.cc

@@ -703,6 +703,10 @@ TEST_F(TitanDBTest, DeleteFilesInRange) {
   ASSERT_TRUE(db_->GetProperty("rocksdb.num-files-at-level6", &value));
   ASSERT_EQ(value, "3");
 
+  std::unique_ptr<Iterator> iter(db_->NewIterator(ReadOptions()));
+  iter->SeekToFirst();
+  ASSERT_TRUE(iter->Valid());
+
   std::string key40 = GenKey(40);
   std::string key80 = GenKey(80);
   Slice start = Slice(key40);
@@ -723,6 +727,18 @@ TEST_F(TitanDBTest, DeleteFilesInRange) {
   // These two files are marked obsolete directly by `DeleteBlobFilesInRanges`
   ASSERT_EQ(blob->NumObsoleteBlobFiles(), 2);
 
+  // The snapshot held by the iterator prevents the blob files from being
+  // purged.
+  ASSERT_OK(db_impl_->TEST_PurgeObsoleteFiles());
+  while (iter->Valid()) {
+    iter->Next();
+    ASSERT_OK(iter->status());
+  }
+  ASSERT_EQ(blob->NumBlobFiles(), 6);
+  ASSERT_EQ(blob->NumObsoleteBlobFiles(), 2);
+
+  // Once the snapshot is released, the blob files should be purged.
+  iter.reset(nullptr);
   ASSERT_OK(db_impl_->TEST_PurgeObsoleteFiles());
   ASSERT_EQ(blob->NumBlobFiles(), 4);
   ASSERT_EQ(blob->NumObsoleteBlobFiles(), 0);

src/version_edit.cc

@@ -69,7 +69,7 @@ Status VersionEdit::DecodeFrom(Slice* src) {
         break;
       case kDeletedBlobFile:
         if (GetVarint64(src, &file_number)) {
-          DeleteBlobFile(file_number);
+          DeleteBlobFile(file_number, 0);
         } else {
           error = "deleted blob file";
         }

src/version_edit.h

@@ -32,8 +32,7 @@ class VersionEdit {
     added_files_.push_back(file);
   }
 
-  void DeleteBlobFile(uint64_t file_number,
-                      SequenceNumber obsolete_sequence = 0) {
+  void DeleteBlobFile(uint64_t file_number, SequenceNumber obsolete_sequence) {
     deleted_files_.emplace_back(std::make_pair(file_number, obsolete_sequence));
   }
 

src/version_test.cc

@@ -134,8 +134,8 @@ TEST_F(VersionTest, VersionEdit) {
   auto file2 = std::make_shared<BlobFileMeta>(5, 6, 0, 0, "", "");
   input.AddBlobFile(file1);
   input.AddBlobFile(file2);
-  input.DeleteBlobFile(7);
-  input.DeleteBlobFile(8);
+  input.DeleteBlobFile(7, 0);
+  input.DeleteBlobFile(8, 0);
   CheckCodec(input);
 }
 
@@ -153,7 +153,7 @@ VersionEdit DeleteBlobFilesEdit(uint32_t cf_id, uint64_t start, uint64_t end) {
   VersionEdit edit;
   edit.SetColumnFamilyID(cf_id);
   for (auto i = start; i < end; i++) {
-    edit.DeleteBlobFile(i);
+    edit.DeleteBlobFile(i, 0);
   }
   return edit;
 }
@@ -367,7 +367,7 @@ TEST_F(VersionTest, BlobFileMetaV1ToV2) {
   VersionEdit edit;
   edit.SetColumnFamilyID(1);
   edit.AddBlobFile(std::make_shared<BlobFileMeta>(1, 1, 0, 0, "", ""));
-  edit.DeleteBlobFile(1);
+  edit.DeleteBlobFile(1, 0);
   edit.AddBlobFile(std::make_shared<BlobFileMeta>(2, 2, 0, 0, "", ""));
   std::string str;
   LegacyEncode(edit, &str);