Fix new CF missing BlobFileSizeCollector (#78)

Summary:
On `CreateColumnFamilies` we should set `BlobFileSizeCollectorFactory`. Fixing it.

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

src/blob_format.cc

@@ -183,7 +183,7 @@ void BlobFileMeta::FileStateTransit(const FileEvent& event) {
       // normal state after flush completed.
       assert(state_ == FileState::kPendingLSM ||
              state_ == FileState::kPendingGC || state_ == FileState::kNormal ||
-             state_ == FileState::kBeingGC);
+             state_ == FileState::kBeingGC || state_ == FileState::kObsolete);
       if (state_ == FileState::kPendingLSM) state_ = FileState::kNormal;
       break;
     case FileEvent::kGCCompleted:

src/blob_gc.cc

@@ -21,7 +21,6 @@ ColumnFamilyData* BlobGC::GetColumnFamilyData() {
 }
 
 void BlobGC::AddOutputFile(BlobFileMeta* blob_file) {
-  blob_file->FileStateTransit(BlobFileMeta::FileEvent::kGCOutput);
   outputs_.push_back(blob_file);
 }
 

src/blob_gc_job.cc

@@ -486,11 +486,12 @@ Status BlobGCJob::InstallOutputBlobFiles() {
                           std::unique_ptr<BlobFileHandle>>>
         files;
     std::string tmp;
-    for (auto& builder : this->blob_file_builders_) {
+    for (auto& builder : blob_file_builders_) {
       auto file = std::make_shared<BlobFileMeta>(
           builder.first->GetNumber(), builder.first->GetFile()->GetFileSize(),
           0, 0, builder.second->GetSmallestKey(),
           builder.second->GetLargestKey());
+      file->FileStateTransit(BlobFileMeta::FileEvent::kGCOutput);
       RecordInHistogram(stats_, TitanStats::GC_OUTPUT_FILE_SIZE,
                         file->file_size());
       if (!tmp.empty()) {
@@ -502,7 +503,7 @@ Status BlobGCJob::InstallOutputBlobFiles() {
     ROCKS_LOG_BUFFER(log_buffer_, "[%s] output[%s]",
                      blob_gc_->column_family_handle()->GetName().c_str(),
                      tmp.c_str());
-    s = this->blob_file_manager_->BatchFinishFiles(
+    s = blob_file_manager_->BatchFinishFiles(
         blob_gc_->column_family_handle()->GetID(), files);
     if (s.ok()) {
       for (auto& file : files) {
@@ -512,7 +513,7 @@ Status BlobGCJob::InstallOutputBlobFiles() {
   } else {
     std::vector<std::unique_ptr<BlobFileHandle>> handles;
     std::string to_delete_files;
-    for (auto& builder : this->blob_file_builders_) {
+    for (auto& builder : blob_file_builders_) {
       if (!to_delete_files.empty()) {
         to_delete_files.append(" ");
       }
@@ -524,7 +525,7 @@ Status BlobGCJob::InstallOutputBlobFiles() {
         "[%s] InstallOutputBlobFiles failed. Delete GC output files: %s",
         blob_gc_->column_family_handle()->GetName().c_str(),
         to_delete_files.c_str());
-    s = this->blob_file_manager_->BatchDeleteFiles(handles);
+    s = blob_file_manager_->BatchDeleteFiles(handles);
   }
   return s;
 }
@@ -532,12 +533,12 @@ Status BlobGCJob::InstallOutputBlobFiles() {
 Status BlobGCJob::RewriteValidKeyToLSM() {
   TitanStopWatch sw(env_, metrics_.gc_update_lsm_micros);
   Status s;
-  auto* db_impl = reinterpret_cast<DBImpl*>(this->base_db_);
+  auto* db_impl = reinterpret_cast<DBImpl*>(base_db_);
 
   WriteOptions wo;
   wo.low_pri = true;
   wo.ignore_missing_column_families = true;
-  for (auto& write_batch : this->rewrite_batches_) {
+  for (auto& write_batch : rewrite_batches_) {
     if (blob_gc_->GetColumnFamilyData()->IsDropped()) {
       s = Status::Aborted("Column family drop");
       break;

src/db_impl.cc

@@ -349,6 +349,8 @@ Status TitanDBImpl::CreateColumnFamilies(
         db_options_, desc.options, blob_manager_, &mutex_, blob_file_set_.get(),
         stats_.get()));
     options.table_factory = titan_table_factory.back();
+    options.table_properties_collector_factories.emplace_back(
+        std::make_shared<BlobFileSizeCollectorFactory>());
     base_descs.emplace_back(desc.name, options);
   }
 

src/titan_db_test.cc

@@ -10,6 +10,7 @@
 
 #include "blob_file_iterator.h"
 #include "blob_file_reader.h"
+#include "blob_file_size_collector.h"
 #include "db_impl.h"
 #include "db_iter.h"
 #include "titan/db.h"
@@ -439,6 +440,17 @@ TEST_F(TitanDBTest, IngestExternalFiles) {
   }
 }
 
+TEST_F(TitanDBTest, NewColumnFamilyHasBlobFileSizeCollector) {
+  Open();
+  AddCF("new_cf");
+  Options opt = db_->GetOptions(cf_handles_.back());
+  ASSERT_EQ(1, opt.table_properties_collector_factories.size());
+  std::unique_ptr<BlobFileSizeCollectorFactory> prop_collector_factory(
+      new BlobFileSizeCollectorFactory());
+  ASSERT_EQ(std::string(prop_collector_factory->Name()),
+            std::string(opt.table_properties_collector_factories[0]->Name()));
+}
+
 TEST_F(TitanDBTest, DropColumnFamily) {
   Open();
   const uint64_t kNumCF = 3;