Remove orphan entry in namespace

(baidu#887)

src/flags.cc

@@ -41,6 +41,7 @@ DEFINE_int32(blockmapping_bucket_num, 19, "Partation num of blockmapping");
 DEFINE_int32(blockmapping_working_thread_num, 5, "Working thread num of blockmapping");
 DEFINE_int32(block_id_allocation_size, 10000, "Block id allocatoin size");
 DEFINE_bool(check_orphan, false, "Check orphan entry in RebuildBlockMap");
+DEFINE_bool(remove_orphan_entry, false, "Remove orphan entry in RebuildBlockMap");
 
 // ha
 DEFINE_string(ha_strategy, "master_slave", "[master_slave, raft, none]");

src/nameserver/namespace.cc

@@ -25,6 +25,7 @@ DECLARE_int32(default_replica_num);
 DECLARE_int32(block_id_allocation_size);
 DECLARE_int32(snapshot_step);
 DECLARE_bool(check_orphan);
+DECLARE_bool(remove_orphan_entry);
 
 const int64_t kRootEntryid = 1;
 
@@ -726,6 +727,18 @@ bool NameSpace::RebuildBlockMap(std::function<void (const FileInfo&)> callback)
             }
         }
         LOG(INFO, "Check orphan done, %lu entries", orphan_entrys.size());
+        if (FLAGS_remove_orphan_entry) {
+            for (std::vector<std::pair<std::string, std::string> >::iterator it =
+                    orphan_entrys.begin(); it != orphan_entrys.end(); ++it) {
+                int64_t entry_id;
+                std::string name;
+                DecodingStoreKey((*it).first, &entry_id, &name);
+                FileInfo file_info;
+                file_info.ParseFromArray((*it).second.data(), (*it).second.size());
+                db_->Delete(leveldb::WriteOptions(), (*it).first);
+                LOG(INFO, "Remove orphan entry E%ld", entry_id);
+            }
+        }
     }
     delete it;
     return true;