[refactor](compaction) submit manual full compaction task to thread pool instead of detached thread (#61222)

Previously, manually triggering full compaction on a single tablet via
HTTP API would create a detached thread for each request, which lacks
concurrency control, deduplication, and metrics tracking.

This PR changes the single-tablet full compaction path to use
`submit_compaction_task()`, submitting to the base compaction thread
pool — consistent with the table-level path and the cloud engine. Base
and cumulative compaction remain unchanged.

Added `force` HTTP parameter to skip permit limiter when submitting full
compaction, allowing compaction to proceed even when resources are
constrained.

### Usage

```bash
# single tablet, default (with permit limiter)
curl -X POST "http://be_host:http_port/api/compaction/run?tablet_id=12345&compact_type=full"

# single tablet, force (skip permit limiter)
curl -X POST "http://be_host:http_port/api/compaction/run?tablet_id=12345&compact_type=full&force=true"

# table level
curl -X POST "http://be_host:http_port/api/compaction/run?table_id=67890&compact_type=full"

# table level, force
curl -X POST "http://be_host:http_port/api/compaction/run?table_id=67890&compact_type=full&force=true"
```

Author: Yukang-Lian

be/src/service/http/action/compaction_action.cpp

@@ -40,7 +40,6 @@
 #include "storage/compaction/cumulative_compaction.h"
 #include "storage/compaction/cumulative_compaction_policy.h"
 #include "storage/compaction/cumulative_compaction_time_series_policy.h"
-#include "storage/compaction/full_compaction.h"
 #include "storage/compaction/single_replica_compaction.h"
 #include "storage/compaction_task_tracker.h"
 #include "storage/olap_define.h"
@@ -150,13 +149,22 @@ Status CompactionAction::_handle_run_compaction(HttpRequest* req, std::string* j
         return Status::NotSupported("The remote = '{}' is not supported", param_remote);
     }
 
+    // "force" = "true" means skip permit limiter when submitting full compaction to thread pool
+    bool force = false;
+    std::string param_force = req->param(PARAM_COMPACTION_FORCE);
+    if (param_force == "true") {
+        force = true;
+    } else if (!param_force.empty() && param_force != "false") {
+        return Status::NotSupported("The force = '{}' is not supported", param_force);
+    }
+
     if (tablet_id == 0 && table_id != 0) {
         std::vector<TabletSharedPtr> tablet_vec = _engine.tablet_manager()->get_all_tablet(
                 [table_id](Tablet* tablet) -> bool { return tablet->get_table_id() == table_id; });
         for (const auto& tablet : tablet_vec) {
             tablet->set_last_full_compaction_schedule_time(UnixMillis());
             RETURN_IF_ERROR(_engine.submit_compaction_task(tablet, CompactionType::FULL_COMPACTION,
-                                                           false, true, 1));
+                                                           force, true, 1));
         }
     } else {
         // 2. fetch the tablet by tablet_id
@@ -178,33 +186,36 @@ Status CompactionAction::_handle_run_compaction(HttpRequest* req, std::string* j
             return Status::OK();
         })
 
-        // 3. execute compaction task
-        std::packaged_task<Status()> task([this, tablet, compaction_type, fetch_from_remote]() {
-            return _execute_compaction_callback(tablet, compaction_type, fetch_from_remote);
-        });
-        std::future<Status> future_obj = task.get_future();
-        std::thread(std::move(task)).detach();
-
-        // 更新schedule_time
-        if (compaction_type == PARAM_COMPACTION_BASE) {
-            tablet->set_last_base_compaction_schedule_time(UnixMillis());
-        } else if (compaction_type == PARAM_COMPACTION_CUMULATIVE) {
-            tablet->set_last_cumu_compaction_schedule_time(UnixMillis());
-        } else if (compaction_type == PARAM_COMPACTION_FULL) {
+        if (compaction_type == PARAM_COMPACTION_FULL) {
+            // 3. submit full compaction task to thread pool
             tablet->set_last_full_compaction_schedule_time(UnixMillis());
-        }
+            RETURN_IF_ERROR(_engine.submit_compaction_task(tablet, CompactionType::FULL_COMPACTION,
+                                                           force, true, 1));
+        } else {
+            // 3. execute base/cumulative compaction task in a detached thread
+            std::packaged_task<Status()> task([this, tablet, compaction_type, fetch_from_remote]() {
+                return _execute_compaction_callback(tablet, compaction_type, fetch_from_remote);
+            });
+            std::future<Status> future_obj = task.get_future();
+            std::thread(std::move(task)).detach();
+
+            if (compaction_type == PARAM_COMPACTION_BASE) {
+                tablet->set_last_base_compaction_schedule_time(UnixMillis());
+            } else if (compaction_type == PARAM_COMPACTION_CUMULATIVE) {
+                tablet->set_last_cumu_compaction_schedule_time(UnixMillis());
+            }
 
-        // 4. wait for result for 2 seconds by async
-        std::future_status status = future_obj.wait_for(std::chrono::seconds(2));
-        if (status == std::future_status::ready) {
-            // fetch execute result
-            Status olap_status = future_obj.get();
-            if (!olap_status.ok()) {
-                return olap_status;
+            // 4. wait for result for 2 seconds by async
+            std::future_status status = future_obj.wait_for(std::chrono::seconds(2));
+            if (status == std::future_status::ready) {
+                Status olap_status = future_obj.get();
+                if (!olap_status.ok()) {
+                    return olap_status;
+                }
+            } else {
+                LOG(INFO) << "Manual compaction task is timeout for waiting "
+                          << (status == std::future_status::timeout);
             }
-        } else {
-            LOG(INFO) << "Manual compaction task is timeout for waiting "
-                      << (status == std::future_status::timeout);
         }
     }
     LOG(INFO) << "Manual compaction task is successfully triggered";
@@ -375,19 +386,6 @@ Status CompactionAction::_execute_compaction_callback(TabletSharedPtr tablet,
                 }
             }
         }
-    } else if (compaction_type == PARAM_COMPACTION_FULL) {
-        FullCompaction full_compaction(_engine, tablet);
-        res = do_compact(full_compaction, CompactionProfileType::FULL);
-        if (!res) {
-            if (res.is<FULL_NO_SUITABLE_VERSION>()) {
-                // Ignore this error code.
-                VLOG_NOTICE << "failed to init full compaction due to no suitable version,"
-                            << "tablet=" << tablet->tablet_id();
-            } else {
-                LOG(WARNING) << "failed to do full compaction. res=" << res
-                             << ", table=" << tablet->tablet_id();
-            }
-        }
     }
     timer.stop();
     LOG(INFO) << "Manual compaction task finish, status=" << res

be/src/service/http/action/compaction_action.h

@@ -41,6 +41,7 @@ const std::string PARAM_COMPACTION_BASE = "base";
 const std::string PARAM_COMPACTION_CUMULATIVE = "cumulative";
 const std::string PARAM_COMPACTION_FULL = "full";
 const std::string PARAM_COMPACTION_REMOTE = "remote";
+const std::string PARAM_COMPACTION_FORCE = "force";
 
 /// This action is used for viewing the compaction status.
 /// See compaction-action.md for details.
@@ -60,9 +61,9 @@ class CompactionAction : public HttpHandlerWithAuth {
     /// param compact_type in req to distinguish the task type, base or cumulative
     Status _handle_run_compaction(HttpRequest* req, std::string* json_result);
 
-    /// thread callback function for the tablet to do compaction
+    /// thread callback function for the tablet to do base/cumulative compaction
     Status _execute_compaction_callback(TabletSharedPtr tablet, const std::string& compaction_type,
-                                        bool fethch_from_remote);
+                                        bool fetch_from_remote);
 
     /// fetch compaction running status
     Status _handle_run_status_compaction(HttpRequest* req, std::string* json_result);

be/test/service/http/compaction_action_test.cpp

@@ -0,0 +1,263 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "service/http/action/compaction_action.h"
+
+#include <event2/http.h>
+#include <gtest/gtest.h>
+
+#include <string>
+
+#include "service/http/http_request.h"
+#include "storage/olap_define.h"
+#include "storage/options.h"
+#include "storage/storage_engine.h"
+#include "util/uid_util.h"
+
+namespace doris {
+
+// Test fixture: creates a minimal StorageEngine with TabletManager initialized.
+// No real tablets are added, so get_tablet() returns nullptr for any id.
+// Tests use -fno-access-control to call private methods directly.
+class CompactionActionTest : public testing::Test {
+public:
+    void SetUp() override {
+        EngineOptions options;
+        options.backend_uid = UniqueId::gen_uid();
+        _storage_engine = std::make_unique<StorageEngine>(options);
+        _evhttp_req = evhttp_request_new(nullptr, nullptr);
+    }
+
+    void TearDown() override {
+        if (_evhttp_req != nullptr) {
+            evhttp_request_free(_evhttp_req);
+        }
+        _storage_engine.reset();
+    }
+
+protected:
+    CompactionAction _make_run_action() {
+        return {CompactionActionType::RUN_COMPACTION, nullptr, *_storage_engine,
+                TPrivilegeHier::GLOBAL, TPrivilegeType::ADMIN};
+    }
+
+    CompactionAction _make_status_action() {
+        return {CompactionActionType::RUN_COMPACTION_STATUS, nullptr, *_storage_engine,
+                TPrivilegeHier::GLOBAL, TPrivilegeType::ADMIN};
+    }
+
+    CompactionAction _make_show_action() {
+        return {CompactionActionType::SHOW_INFO, nullptr, *_storage_engine, TPrivilegeHier::GLOBAL,
+                TPrivilegeType::ADMIN};
+    }
+
+    evhttp_request* _evhttp_req = nullptr;
+    std::unique_ptr<StorageEngine> _storage_engine;
+};
+
+// ==================== _handle_run_compaction tests ====================
+
+TEST_F(CompactionActionTest, RunCompactionMissingBothIds) {
+    auto action = _make_run_action();
+    HttpRequest req(_evhttp_req);
+    req._params[PARAM_COMPACTION_TYPE] = PARAM_COMPACTION_FULL;
+
+    std::string json_result;
+    Status st = action._handle_run_compaction(&req, &json_result);
+    EXPECT_FALSE(st.ok());
+    EXPECT_TRUE(st.to_string().find("can not be empty at the same time") != std::string::npos);
+}
+
+TEST_F(CompactionActionTest, RunCompactionBothIdsSet) {
+    auto action = _make_run_action();
+    HttpRequest req(_evhttp_req);
+    req._params[TABLET_ID_KEY] = "12345";
+    req._params[TABLE_ID_KEY] = "67890";
+    req._params[PARAM_COMPACTION_TYPE] = PARAM_COMPACTION_FULL;
+
+    std::string json_result;
+    Status st = action._handle_run_compaction(&req, &json_result);
+    EXPECT_FALSE(st.ok());
+    EXPECT_TRUE(st.to_string().find("can not be set at the same time") != std::string::npos);
+}
+
+TEST_F(CompactionActionTest, RunCompactionInvalidType) {
+    auto action = _make_run_action();
+    HttpRequest req(_evhttp_req);
+    req._params[TABLET_ID_KEY] = "12345";
+    req._params[PARAM_COMPACTION_TYPE] = "invalid_type";
+
+    std::string json_result;
+    Status st = action._handle_run_compaction(&req, &json_result);
+    EXPECT_FALSE(st.ok());
+    EXPECT_TRUE(st.to_string().find("not supported") != std::string::npos);
+}
+
+TEST_F(CompactionActionTest, RunCompactionInvalidRemoteParam) {
+    auto action = _make_run_action();
+    HttpRequest req(_evhttp_req);
+    req._params[TABLET_ID_KEY] = "12345";
+    req._params[PARAM_COMPACTION_TYPE] = PARAM_COMPACTION_FULL;
+    req._params[PARAM_COMPACTION_REMOTE] = "invalid_value";
+
+    std::string json_result;
+    Status st = action._handle_run_compaction(&req, &json_result);
+    EXPECT_FALSE(st.ok());
+    EXPECT_TRUE(st.to_string().find("not supported") != std::string::npos);
+}
+
+TEST_F(CompactionActionTest, RunCompactionInvalidTabletId) {
+    auto action = _make_run_action();
+    HttpRequest req(_evhttp_req);
+    req._params[TABLET_ID_KEY] = "not_a_number";
+    req._params[PARAM_COMPACTION_TYPE] = PARAM_COMPACTION_FULL;
+
+    std::string json_result;
+    Status st = action._handle_run_compaction(&req, &json_result);
+    EXPECT_FALSE(st.ok());
+    EXPECT_TRUE(st.to_string().find("convert") != std::string::npos);
+}
+
+TEST_F(CompactionActionTest, RunFullCompactionTabletNotFound) {
+    auto action = _make_run_action();
+    HttpRequest req(_evhttp_req);
+    req._params[TABLET_ID_KEY] = "99999";
+    req._params[PARAM_COMPACTION_TYPE] = PARAM_COMPACTION_FULL;
+
+    std::string json_result;
+    Status st = action._handle_run_compaction(&req, &json_result);
+    EXPECT_TRUE(st.is<ErrorCode::NOT_FOUND>());
+}
+
+TEST_F(CompactionActionTest, RunBaseCompactionTabletNotFound) {
+    auto action = _make_run_action();
+    HttpRequest req(_evhttp_req);
+    req._params[TABLET_ID_KEY] = "99999";
+    req._params[PARAM_COMPACTION_TYPE] = PARAM_COMPACTION_BASE;
+
+    std::string json_result;
+    Status st = action._handle_run_compaction(&req, &json_result);
+    EXPECT_TRUE(st.is<ErrorCode::NOT_FOUND>());
+}
+
+TEST_F(CompactionActionTest, RunCumulativeCompactionTabletNotFound) {
+    auto action = _make_run_action();
+    HttpRequest req(_evhttp_req);
+    req._params[TABLET_ID_KEY] = "99999";
+    req._params[PARAM_COMPACTION_TYPE] = PARAM_COMPACTION_CUMULATIVE;
+
+    std::string json_result;
+    Status st = action._handle_run_compaction(&req, &json_result);
+    EXPECT_TRUE(st.is<ErrorCode::NOT_FOUND>());
+}
+
+// ==================== force parameter tests ====================
+
+TEST_F(CompactionActionTest, RunCompactionInvalidForceParam) {
+    auto action = _make_run_action();
+    HttpRequest req(_evhttp_req);
+    req._params[TABLET_ID_KEY] = "12345";
+    req._params[PARAM_COMPACTION_TYPE] = PARAM_COMPACTION_FULL;
+    req._params[PARAM_COMPACTION_FORCE] = "invalid_value";
+
+    std::string json_result;
+    Status st = action._handle_run_compaction(&req, &json_result);
+    EXPECT_FALSE(st.ok());
+    EXPECT_TRUE(st.to_string().find("not supported") != std::string::npos);
+}
+
+TEST_F(CompactionActionTest, RunFullCompactionForceTrue) {
+    auto action = _make_run_action();
+    HttpRequest req(_evhttp_req);
+    req._params[TABLET_ID_KEY] = "99999";
+    req._params[PARAM_COMPACTION_TYPE] = PARAM_COMPACTION_FULL;
+    req._params[PARAM_COMPACTION_FORCE] = "true";
+
+    std::string json_result;
+    Status st = action._handle_run_compaction(&req, &json_result);
+    // tablet not found, but force param was parsed successfully
+    EXPECT_TRUE(st.is<ErrorCode::NOT_FOUND>());
+}
+
+TEST_F(CompactionActionTest, RunFullCompactionForceFalse) {
+    auto action = _make_run_action();
+    HttpRequest req(_evhttp_req);
+    req._params[TABLET_ID_KEY] = "99999";
+    req._params[PARAM_COMPACTION_TYPE] = PARAM_COMPACTION_FULL;
+    req._params[PARAM_COMPACTION_FORCE] = "false";
+
+    std::string json_result;
+    Status st = action._handle_run_compaction(&req, &json_result);
+    EXPECT_TRUE(st.is<ErrorCode::NOT_FOUND>());
+}
+
+TEST_F(CompactionActionTest, RunCompactionForceWithTableId) {
+    auto action = _make_run_action();
+    HttpRequest req(_evhttp_req);
+    req._params[TABLE_ID_KEY] = "99999";
+    req._params[PARAM_COMPACTION_TYPE] = PARAM_COMPACTION_FULL;
+    req._params[PARAM_COMPACTION_FORCE] = "true";
+
+    std::string json_result;
+    // table_id path with no matching tablets returns success (empty loop)
+    Status st = action._handle_run_compaction(&req, &json_result);
+    EXPECT_TRUE(st.ok());
+}
+
+// ==================== _handle_show_compaction tests ====================
+
+TEST_F(CompactionActionTest, ShowCompactionMissingTabletId) {
+    auto action = _make_show_action();
+    HttpRequest req(_evhttp_req);
+
+    std::string json_result;
+    Status st = action._handle_show_compaction(&req, &json_result);
+    EXPECT_FALSE(st.ok());
+}
+
+TEST_F(CompactionActionTest, ShowCompactionTabletNotFound) {
+    auto action = _make_show_action();
+    HttpRequest req(_evhttp_req);
+    req._params[TABLET_ID_KEY] = "99999";
+
+    std::string json_result;
+    Status st = action._handle_show_compaction(&req, &json_result);
+    EXPECT_TRUE(st.is<ErrorCode::NOT_FOUND>());
+}
+
+// ==================== _handle_run_status_compaction tests ====================
+
+TEST_F(CompactionActionTest, RunStatusCompactionOverall) {
+    auto action = _make_status_action();
+    HttpRequest req(_evhttp_req);
+    // tablet_id not set → returns overall compaction status
+    std::string json_result;
+    Status st = action._handle_run_status_compaction(&req, &json_result);
+    EXPECT_TRUE(st.ok());
+}
+
+TEST_F(CompactionActionTest, RunStatusCompactionTabletNotFound) {
+    auto action = _make_status_action();
+    HttpRequest req(_evhttp_req);
+    req._params[TABLET_ID_KEY] = "99999";
+
+    std::string json_result;
+    Status st = action._handle_run_status_compaction(&req, &json_result);
+    EXPECT_FALSE(st.ok());
+}
+
+} // namespace doris