[BugFix] Fix disable base compaction with minute granularity & fe/be …

…recover

Signed-off-by: meegoo <meegoo.sr@gmail.com>

be/src/storage/compaction_manager.cpp

@@ -162,6 +162,9 @@ void CompactionManager::update_candidates(std::vector<CompactionCandidate> candi
             }
         }
         for (auto& candidate : candidates) {
+            if (_check_compaction_disabled(candidate)) {
+                continue;
+            }
             if (candidate.tablet->enable_compaction()) {
                 VLOG(2) << "update candidate " << candidate.tablet->tablet_id() << " type "
                         << starrocks::to_string(candidate.type) << " score " << candidate.score;
@@ -200,6 +203,25 @@ void CompactionManager::remove_candidate(int64_t tablet_id) {
     }
 }
 
+bool CompactionManager::_check_compaction_disabled(const CompactionCandidate& candidate) {
+    if (candidate.type == CompactionType::BASE_COMPACTION &&
+        _table_to_disable_deadline_map.find(candidate.tablet->tablet_meta()->table_id()) !=
+                _table_to_disable_deadline_map.end()) {
+        int64_t deadline = _table_to_disable_deadline_map[candidate.tablet->tablet_meta()->table_id()];
+        if (deadline > 0 && UnixSeconds() < deadline) {
+            return true;
+        } else {
+            // disable compaction deadline has passed, remove it from map
+            _table_to_disable_deadline_map.erase(candidate.tablet->tablet_meta()->table_id());
+            // check if the tablet should compact now after the deadline
+            update_tablet_async(candidate.tablet);
+            LOG(INFO) << "remove disable table compaction, table_id:" << candidate.tablet->tablet_meta()->table_id()
+                      << ", deadline:" << deadline;
+        }
+    }
+    return false;
+}
+
 bool CompactionManager::_check_precondition(const CompactionCandidate& candidate) {
     if (!candidate.tablet) {
         LOG(WARNING) << "candidate with null tablet";
@@ -212,19 +234,6 @@ bool CompactionManager::_check_precondition(const CompactionCandidate& candidate
         return false;
     }
 
-    // check if the table base compaction is disabled
-    if (candidate.type == CompactionType::BASE_COMPACTION &&
-        _table_to_disable_deadline_map.find(tablet->tablet_meta()->table_id()) !=
-                _table_to_disable_deadline_map.end()) {
-        int64_t deadline = _table_to_disable_deadline_map[tablet->tablet_meta()->table_id()];
-        if (deadline > 0 && UnixSeconds() < deadline) {
-            VLOG(2) << "skip tablet:" << tablet->tablet_id() << " because table is disabled";
-            return false;
-        } else {
-            _table_to_disable_deadline_map.erase(tablet->tablet_meta()->table_id());
-        }
-    }
-
     int64_t last_failure_ts = 0;
     DataDir* data_dir = tablet->data_dir();
     if (candidate.type == CUMULATIVE_COMPACTION) {
@@ -291,6 +300,10 @@ bool CompactionManager::pick_candidate(CompactionCandidate* candidate) {
 
     auto iter = _compaction_candidates.begin();
     while (iter != _compaction_candidates.end()) {
+        if (_check_compaction_disabled(*iter)) {
+            _compaction_candidates.erase(iter++);
+            continue;
+        }
         if (_check_precondition(*iter)) {
             *candidate = *iter;
             _compaction_candidates.erase(iter);
@@ -654,8 +667,11 @@ int CompactionManager::get_waiting_task_num() {
 
 void CompactionManager::disable_table_compaction(int64_t table_id, int64_t deadline) {
     std::lock_guard lg(_candidates_mutex);
-    VLOG(2) << "disable table compaction, table_id:" << table_id << ", deadline:" << deadline;
+    if (_table_to_disable_deadline_map.find(table_id) == _table_to_disable_deadline_map.end()) {
+        LOG(INFO) << "start disable table compaction, table_id:" << table_id << ", deadline:" << deadline;
+    }
     _table_to_disable_deadline_map[table_id] = deadline;
+    VLOG(2) << "disable table compaction, table_id:" << table_id << ", deadline:" << deadline;
 }
 
 } // namespace starrocks

be/src/storage/compaction_manager.h

@@ -146,6 +146,7 @@ class CompactionManager {
     CompactionManager& operator=(CompactionManager&& compaction_manager) = delete;
 
     void _dispatch_worker();
+    bool _check_compaction_disabled(const CompactionCandidate& candidate);
     bool _check_precondition(const CompactionCandidate& candidate);
     void _schedule();
     void _notify();

be/src/storage/compaction_task.cpp

@@ -143,6 +143,8 @@ void CompactionTask::run() {
                                           _task_info.to_string());
     if (!status.ok()) {
         LOG(WARNING) << msg;
+    } else {
+        LOG(INFO) << msg;
     }
 }
 

be/src/storage/tablet_meta.h

@@ -237,6 +237,9 @@ class TabletMeta {
 
     const TabletSchemaCSPtr& source_schema() const { return _source_schema; }
 
+    // for test
+    void TEST_set_table_id(int64_t table_id);
+
 private:
     int64_t _mem_usage() const { return sizeof(TabletMeta); }
 
@@ -312,6 +315,10 @@ inline int64_t TabletMeta::table_id() const {
     return _table_id;
 }
 
+inline void TabletMeta::TEST_set_table_id(int64_t table_id) {
+    _table_id = table_id;
+}
+
 inline int64_t TabletMeta::partition_id() const {
     return _partition_id;
 }

be/test/storage/compaction_manager_test.cpp

@@ -188,13 +188,15 @@ TEST_F(CompactionManagerTest, test_disable_compaction) {
         TabletSharedPtr tablet = std::make_shared<Tablet>();
         TabletMetaSharedPtr tablet_meta = std::make_shared<TabletMeta>();
         tablet_meta->set_tablet_id(i);
+        tablet_meta->TEST_set_table_id(1);
         tablet->set_tablet_meta(tablet_meta);
         tablet->set_data_dir(&data_dir);
         tablet->set_tablet_state(TABLET_RUNNING);
 
         CompactionCandidate candidate;
         candidate.tablet = tablet;
         candidate.score = i;
+        candidate.type = BASE_COMPACTION;
         candidates.push_back(candidate);
     }
 
@@ -204,9 +206,9 @@ TEST_F(CompactionManagerTest, test_disable_compaction) {
 
     _engine->compaction_manager()->update_candidates(candidates);
 
-    {
-        ASSERT_EQ(10, _engine->compaction_manager()->candidates_size());
+    _engine->compaction_manager()->disable_table_compaction(1, UnixSeconds() + 5);
 
+    {
         int64_t valid_condidates = 0;
         while (true) {
             CompactionCandidate candidate;
@@ -216,10 +218,38 @@ TEST_F(CompactionManagerTest, test_disable_compaction) {
             }
             ++valid_condidates;
         }
-        ASSERT_EQ(10, valid_condidates);
+        ASSERT_EQ(0, valid_condidates);
     }
+}
 
-    _engine->compaction_manager()->disable_table_compaction(0, UnixSeconds() + 3600);
+TEST_F(CompactionManagerTest, test_remove_disable_compaction) {
+    std::vector<CompactionCandidate> candidates;
+    DataDir data_dir("./data_dir");
+    for (int i = 0; i < 10; i++) {
+        TabletSharedPtr tablet = std::make_shared<Tablet>();
+        TabletMetaSharedPtr tablet_meta = std::make_shared<TabletMeta>();
+        tablet_meta->set_tablet_id(i);
+        tablet_meta->TEST_set_table_id(2);
+        tablet->set_tablet_meta(tablet_meta);
+        tablet->set_data_dir(&data_dir);
+        tablet->set_tablet_state(TABLET_RUNNING);
+
+        CompactionCandidate candidate;
+        candidate.tablet = tablet;
+        candidate.score = i;
+        candidate.type = BASE_COMPACTION;
+        candidates.push_back(candidate);
+    }
+
+    std::random_device rd;
+    std::mt19937 g(rd());
+    std::shuffle(candidates.begin(), candidates.end(), g);
+
+    _engine->compaction_manager()->update_candidates(candidates);
+
+    _engine->compaction_manager()->disable_table_compaction(2, UnixSeconds());
+
+    sleep(1);
 
     {
         int64_t valid_condidates = 0;
@@ -231,7 +261,7 @@ TEST_F(CompactionManagerTest, test_disable_compaction) {
             }
             ++valid_condidates;
         }
-        ASSERT_EQ(0, valid_condidates);
+        ASSERT_EQ(10, valid_condidates);
     }
 }
 

fe/fe-core/src/main/java/com/starrocks/alter/AlterJobExecutor.java

@@ -387,8 +387,9 @@ public Void visitModifyTablePropertiesClause(ModifyTablePropertiesClause clause,
                     schemaChangeHandler.updateTableMeta(db, tableName.getTbl(), properties,
                             TTabletMetaType.BASE_COMPACTION_FORBIDDEN_TIME_RANGES);
                 } catch (Exception e) {
-                    LOG.warn("Failed to update base compaction forbidden time ranges: ", e);
-                    throw new DdlException("Failed to update base compaction forbidden time ranges: " + e.getMessage());
+                    LOG.warn("Failed to update base compaction forbidden time ranges: " + tableName.getTbl(), e);
+                    throw new DdlException("Failed to update base compaction forbidden time ranges for "
+                            + tableName.getTbl() + ": " + e.getMessage());
                 }
             } else if (properties.containsKey(PropertyAnalyzer.PROPERTIES_BINLOG_ENABLE) ||
                     properties.containsKey(PropertyAnalyzer.PROPERTIES_BINLOG_TTL) ||

fe/fe-core/src/main/java/com/starrocks/catalog/OlapTable.java

@@ -2655,6 +2655,18 @@ public void setBaseCompactionForbiddenTimeRanges(String baseCompactionForbiddenT
         tableProperty.buildBaseCompactionForbiddenTimeRanges();
     }
 
+    public void updateBaseCompactionForbiddenTimeRanges(boolean isDrop) {
+        try {
+            if (isDrop && getBaseCompactionForbiddenTimeRanges().isEmpty()) {
+                return;
+            }
+            GlobalStateMgr.getCurrentState().getCompactionControlScheduler().updateTableForbiddenTimeRanges(
+                    getId(), isDrop ? "" : getBaseCompactionForbiddenTimeRanges());
+        } catch (Exception e) {
+            LOG.warn("Failed to update base compaction forbidden time ranges for " + getName(), e);
+        }
+    }
+
     public TWriteQuorumType writeQuorum() {
         if (tableProperty != null) {
             return tableProperty.writeQuorum();
@@ -3126,6 +3138,7 @@ public void onReload() {
         analyzePartitionInfo();
         analyzeRollupIndexMeta();
         tryToAssignIndexId();
+        updateBaseCompactionForbiddenTimeRanges(false);
 
         // register constraints from global state manager
         GlobalConstraintManager globalConstraintManager = GlobalStateMgr.getCurrentState().getGlobalConstraintManager();
@@ -3264,6 +3277,8 @@ public void onDrop(Database db, boolean force, boolean replay) {
             sendDropAutoIncrementMapTask();
         }
 
+        updateBaseCompactionForbiddenTimeRanges(true);
+
         // unregister constraints from global state manager
         GlobalConstraintManager globalConstraintManager = GlobalStateMgr.getCurrentState().getGlobalConstraintManager();
         globalConstraintManager.unRegisterConstraint(this);