Fix fallback mode fail when it encounter GC-ed file (#30)

Summary:
In fallback mode, table builder will try to read value from blob file and write the value back to LSM. If the corresponding blob file had been GC-ed and deleted, however, table builder will return error and fail compaction. Fixing it by writing the blob index to output if table builder get error when reading blob value.

Test Plan:
See the new test.
Signed-off-by: Yi Wu <yiwu@pingcap.com>

src/db_impl.h

@@ -88,6 +88,9 @@ class TitanDBImpl : public TitanDB {
 
   void StartBackgroundTasks();
 
+  Status TEST_StartGC(uint32_t column_family_id);
+  Status TEST_PurgeObsoleteFiles();
+
  private:
   class FileManager;
   friend class FileManager;
@@ -128,9 +131,9 @@ class TitanDBImpl : public TitanDB {
   static void BGWorkGC(void* db);
   void BackgroundCallGC();
   Status BackgroundGC(LogBuffer* log_buffer);
-  Status TEST_StartGC(uint32_t column_family_id);
 
   void PurgeObsoleteFiles();
+  Status PurgeObsoleteFilesImpl();
 
   SequenceNumber GetOldestSnapshotSequence() {
     SequenceNumber oldest_snapshot = kMaxSequenceNumber;

src/db_impl_files.cc

@@ -3,7 +3,7 @@
 namespace rocksdb {
 namespace titandb {
 
-void TitanDBImpl::PurgeObsoleteFiles() {
+Status TitanDBImpl::PurgeObsoleteFilesImpl() {
   Status s;
   std::vector<std::string> candidate_files;
   auto oldest_sequence = GetOldestSnapshotSequence();
@@ -22,13 +22,21 @@ void TitanDBImpl::PurgeObsoleteFiles() {
   for (const auto& candidate_file : candidate_files) {
     ROCKS_LOG_INFO(db_options_.info_log, "Titan deleting obsolete file [%s]",
                    candidate_file.c_str());
-    s = env_->DeleteFile(candidate_file);
+    Status delete_status = env_->DeleteFile(candidate_file);
     if (!s.ok()) {
-      fprintf(stderr, "Titan deleting file [%s] failed, status:%s",
-              candidate_file.c_str(), s.ToString().c_str());
-      abort();
+      s = delete_status;
     }
   }
+  return s;
+}
+
+void TitanDBImpl::PurgeObsoleteFiles() {
+  Status s __attribute__((__unused__)) = PurgeObsoleteFilesImpl();
+  assert(s.ok());
+}
+
+Status TitanDBImpl::TEST_PurgeObsoleteFiles() {
+  return PurgeObsoleteFilesImpl();
 }
 
 }  // namespace titandb

src/table_builder.cc

@@ -29,12 +29,18 @@ void TitanTableBuilder::Add(const Slice& key, const Slice& value) {
     assert(storage != nullptr);
 
     ReadOptions options;  // dummy option
-    status_ = storage->Get(options, index, &record, &buffer);
-    if (ok()) {
+    Status get_status = storage->Get(options, index, &record, &buffer);
+    if (get_status.ok()) {
       ikey.type = kTypeValue;
       std::string index_key;
       AppendInternalKey(&index_key, ikey);
       base_builder_->Add(index_key, record.value);
+    } else {
+      // Get blob value can fail if corresponding blob file has been GC-ed
+      // deleted. In this case we write the blob index as is to compaction
+      // output.
+      // TODO: return error if it is indeed an error.
+      base_builder_->Add(key, value);
     }
   } else if (ikey.type == kTypeValue &&
              value.size() >= cf_options_.min_blob_size &&

src/titan_db_test.cc

@@ -697,6 +697,32 @@ TEST_F(TitanDBTest, BlobRunModeBasic) {
   version.clear();
 }
 
+TEST_F(TitanDBTest, FallbackModeEncounterMissingBlobFile) {
+  options_.disable_background_gc = true;
+  options_.blob_file_discardable_ratio = 0.01;
+  options_.min_blob_size = true;
+  Open();
+  ASSERT_OK(db_->Put(WriteOptions(), "foo", "v1"));
+  ASSERT_OK(db_->Put(WriteOptions(), "bar", "v1"));
+  ASSERT_OK(db_->Flush(FlushOptions()));
+  ASSERT_EQ(1, GetBlobStorage().lock()->NumBlobFiles());
+  ASSERT_OK(db_->Delete(WriteOptions(), "foo"));
+  ASSERT_OK(db_->Flush(FlushOptions()));
+  uint32_t default_cf_id = db_->DefaultColumnFamily()->GetID();
+  // GC the first blob file.
+  ASSERT_OK(db_impl_->TEST_StartGC(default_cf_id));
+  ASSERT_EQ(2, GetBlobStorage().lock()->NumBlobFiles());
+  ASSERT_OK(db_impl_->TEST_PurgeObsoleteFiles());
+  ASSERT_EQ(1, GetBlobStorage().lock()->NumBlobFiles());
+  ASSERT_OK(db_->SetOptions({{"blob_run_mode", "kFallback"}}));
+  // Run compaction in fallback mode. Make sure it correctly handle the
+  // missing blob file.
+  Slice begin("foo");
+  Slice end("foo1");
+  ASSERT_OK(db_->CompactRange(CompactRangeOptions(), &begin, &end));
+  VerifyDB({{"bar", "v1"}});
+}
+
 }  // namespace titandb
 }  // namespace rocksdb