add garbase collect by rowsetid

be/src/common/config.h

@@ -415,7 +415,7 @@ namespace config {
     // path gc
     CONF_Bool(path_gc_check, "true");
     CONF_Int32(path_gc_check_interval_second, "1800");
-    CONF_Int32(path_gc_check_step, "-1");
+    CONF_Int32(path_gc_check_step, "1000");
     CONF_Int32(path_gc_check_step_interval_ms, "10");
     CONF_Int32(path_scan_interval_second, "1800");
 } // namespace config

be/src/olap/data_dir.cpp

@@ -932,6 +932,52 @@ void DataDir::perform_path_gc() {
     LOG(INFO) << "finished one time path gc.";
 }
 
+void DataDir::perform_path_gc_by_rowsetid() {
+    // init the set of valid path
+    // validate the path in data dir
+    std::unique_lock<std::mutex> lck(_check_path_mutex);
+    cv.wait(lck, [this]{return _all_check_paths.size() > 0;});
+    LOG(INFO) << "start to path gc by rowsetid.";
+    int counter = 0;
+    for (auto& path : _all_check_paths) {
+        ++counter;
+        if (config::path_gc_check_step > 0 && counter % config::path_gc_check_step == 0) {
+            usleep(config::path_gc_check_step_interval_ms * 1000);
+        }
+        TTabletId tablet_id = -1;
+        TSchemaHash schema_hash = -1;
+        bool is_valid = _tablet_manager->get_tablet_id_and_schema_hash_from_path(path,
+                &tablet_id, &schema_hash);
+        if (!is_valid) {
+            LOG(WARNING) << "unknown path:" << path;
+            continue;
+        }
+        if (tablet_id > 0 && schema_hash > 0) {
+            // tablet schema hash path or rowset file path
+            // gc thread should get tablet include deleted tablet
+            // or it will delete rowset file before tablet is garbage collected
+            RowsetId rowset_id = -1;
+            bool is_rowset_file = _tablet_manager->get_rowset_id_from_path(path, &rowset_id);
+            if (is_rowset_file) {
+                TabletSharedPtr tablet = _tablet_manager->get_tablet(tablet_id, schema_hash, true);
+                if (tablet != nullptr) {
+                    bool valid = tablet->check_rowset_id(rowset_id);
+                    if (!valid) {
+                        // if the rowset id is less than tablet's initial end rowset id
+                        // and the path is not in unused_rowsets, delete the path.
+                        if (rowset_id < tablet->initial_end_rowset_id()
+                                && !StorageEngine::instance()->check_rowset_id_in_unused_rowsets(rowset_id)) {
+                            _process_garbage_path(path);
+                        }
+                    }
+                }
+            }
+        }
+    }
+    _all_check_paths.clear();
+    LOG(INFO) << "finished one time path gc by rowsetid.";
+}
+
 // path producer
 void DataDir::perform_path_scan() {
     {

be/src/olap/data_dir.h

@@ -113,6 +113,7 @@ class DataDir {
     // this function will collect garbage paths scaned by last function
     void perform_path_gc();
 
+    void perform_path_gc_by_rowsetid();
 
     OLAPStatus remove_old_meta_and_files();
 

be/src/olap/storage_engine.cpp

@@ -944,6 +944,19 @@ OLAPStatus StorageEngine::execute_task(EngineTask* task) {
     }
 }
 
+// check whether any unused rowsets's id equal to rowset_id
+bool StorageEngine::check_rowset_id_in_unused_rowsets(RowsetId rowset_id) {
+    _gc_mutex.lock();
+    for (auto& _unused_rowset_pair : _unused_rowsets) {
+        if (_unused_rowset_pair.second->rowset_id() == rowset_id) {
+            _gc_mutex.unlock();
+            return true;
+        }
+    }
+    _gc_mutex.unlock();
+    return false;
+}
+
 void* StorageEngine::_path_gc_thread_callback(void* arg) {
 #ifdef GOOGLE_PROFILER
     ProfilerRegisterThread();
@@ -959,12 +972,12 @@ void* StorageEngine::_path_gc_thread_callback(void* arg) {
 
     while (true) {
         LOG(INFO) << "try to perform path gc!";
-        // TODO(ygl): stop gc temp because could not define all pending dirs currently
-        // ((DataDir*)arg)->perform_path_gc();
+        // perform path gc by rowset id
+        ((DataDir*)arg)->perform_path_gc_by_rowsetid();
         usleep(interval * 1000000);
     }
 
-    return NULL;
+    return nullptr;
 }
 
 void* StorageEngine::_path_scan_thread_callback(void* arg) {
@@ -986,7 +999,7 @@ void* StorageEngine::_path_scan_thread_callback(void* arg) {
         usleep(interval * 1000000);
     }
 
-    return NULL;
+    return nullptr;
 }
 
 }  // namespace doris

be/src/olap/storage_engine.h

@@ -193,6 +193,8 @@ class StorageEngine {
     TabletManager* tablet_manager() { return _tablet_manager.get(); }
     TxnManager* txn_manager() { return _txn_manager.get(); }
 
+    bool check_rowset_id_in_unused_rowsets(RowsetId rowset_id);
+
 private:
     OLAPStatus check_all_root_path_cluster_id();
 

be/src/olap/tablet.cpp

@@ -839,6 +839,28 @@ bool Tablet::check_path(const std::string& path_to_check) {
             return true;
         }
     }
+    for (auto& inc_version_rowset : _inc_rs_version_map) {
+        bool ret = inc_version_rowset.second->check_path(path_to_check);
+        if (ret) {
+            return true;
+        }
+    }
+    return false;
+}
+
+bool Tablet::check_rowset_id(RowsetId rowset_id) {
+    ReadLock rdlock(&_meta_lock);
+    for (auto& version_rowset : _rs_version_map) {
+        if (version_rowset.second->rowset_id() == rowset_id) {
+            return true;
+        }
+    }
+
+    for (auto& inc_version_rowset : _inc_rs_version_map) {
+        if (inc_version_rowset.second->rowset_id() == rowset_id) {
+            return true;
+        }
+    }
     return false;
 }
 

be/src/olap/tablet.h

@@ -219,11 +219,18 @@ class Tablet : public std::enable_shared_from_this<Tablet> {
 
     bool check_path(const std::string& check_path);
 
+    // check rowset_id is valid
+    bool check_rowset_id(RowsetId rowset_id);
+
     OLAPStatus next_rowset_id(RowsetId* id);
     OLAPStatus set_next_rowset_id(RowsetId new_rowset_id);
 
     OLAPStatus set_partition_id(int64_t partition_id);
 
+    RowsetId initial_end_rowset_id() {
+        return _tablet_meta->initial_end_rowset_id();
+    }
+
 private:
     void _print_missed_versions(const std::vector<Version>& missed_versions) const;
     OLAPStatus _check_added_rowset(const RowsetSharedPtr& rowset);

be/src/olap/tablet_meta.cpp

@@ -339,6 +339,7 @@ OLAPStatus TabletMeta::init_from_pb(const TabletMetaPB& tablet_meta_pb) {
     _cumulative_layer_point = tablet_meta_pb.cumulative_layer_point();
     _tablet_uid = TabletUid(tablet_meta_pb.tablet_uid());
     _end_rowset_id = tablet_meta_pb.end_rowset_id();
+    _initial_end_rowset_id = tablet_meta_pb.end_rowset_id();
     _next_rowset_id = _end_rowset_id + 1;
 
     // init _tablet_state

be/src/olap/tablet_meta.h

@@ -191,6 +191,10 @@ class TabletMeta {
 
     OLAPStatus set_partition_id(int64_t partition_id);
 
+    RowsetId initial_end_rowset_id() {
+        return _initial_end_rowset_id;
+    }
+
 private:
     OLAPStatus _save_meta(DataDir* data_dir);
 
@@ -205,6 +209,7 @@ class TabletMeta {
     TabletUid _tablet_uid;
     RowsetId _next_rowset_id = 10000;
     RowsetId _end_rowset_id;
+    RowsetId _initial_end_rowset_id;
     RowsetId _batch_interval = 10000;