This is an automated email from the ASF dual-hosted git repository. yiguolei pushed a commit to branch branch-4.2 in repository https://gitbox.apache.org/repos/asf/doris.git
commit bf8a9f56feb9797036b4c8d8f654cb62deb51614 Author: Jamie <[email protected]> AuthorDate: Thu Sep 10 07:33:53 2026 +0800 branch-4.1: [fix](compaction): avoid decrementing uncounted cumulative compaction threads #66752 (#67662) ### What problem does this PR solve? Issue Number: None Related PR: #66752 Problem Summary: Backport the cumulative compaction thread-accounting fix from #66752 to `branch-4.1`. When tablet global compaction lock acquisition fails, the worker exits before `_cumu_compaction_thread_pool_used_threads` is incremented. The deferred cleanup previously decremented the counter unconditionally, which could make it negative and distort compaction scheduling capacity. This backport preserves the 4.1 compaction structure and guards cleanup so the counter is decremented only after it has actually been incremented. It also carries the focused C++ regression test and the Cloud fault-injection regression suite from the source fix. Local validation: - `git diff --check` - Clang Format 16 dry-run for the changed C++ files - `clang++ -fsyntax-only` for `cloud_storage_engine.cpp` and `cloud_compaction_test.cpp` using an existing 4.1 ASAN UT compile database - Groovy 4.0.19 parse for `test_cloud_cumu_compaction_global_lock_thread_count.groovy` The focused C++ and Docker regression tests were not executed locally; hosted CI remains required. ### Release note Fix Cloud cumulative compaction thread accounting after global lock acquisition failures. ### Check List (For Author) - Test <!-- At least one of them must be included. --> - [x] Regression test - [x] Unit Test - [ ] Manual test (add detailed scripts or steps below) - [ ] No need to test or manual test. Explain why: - [ ] This is a refactor/code format and no logic has been changed. - [ ] Previous test can cover this change. - [ ] No code files have been changed. - [ ] Other reason <!-- Add your reason? --> Test cases: - `CloudCompactionTest.cumulative_global_lock_failure_keeps_thread_count_balanced` - `test_cloud_cumu_compaction_global_lock_thread_count` Signed-off-by: Yukang-Lian <[email protected]> Co-authored-by: dzr171712 <[email protected]> --- be/src/cloud/cloud_storage_engine.cpp | 12 +- be/test/cloud/cloud_compaction_test.cpp | 53 ++++++ ...cumu_compaction_global_lock_thread_count.groovy | 186 +++++++++++++++++++++ 3 files changed, 247 insertions(+), 4 deletions(-) diff --git a/be/src/cloud/cloud_storage_engine.cpp b/be/src/cloud/cloud_storage_engine.cpp index 1727b44720c..2b56109b702 100644 --- a/be/src/cloud/cloud_storage_engine.cpp +++ b/be/src/cloud/cloud_storage_engine.cpp @@ -1021,15 +1021,18 @@ Status CloudStorageEngine::_submit_cumulative_compaction_task(const CloudTabletS signal::tablet_id = tablet->tablet_id(); g_cumu_compaction_running_task_count << 1; bool is_large_task = true; + bool cumu_thread_counted = false; Defer defer {[&]() { DBUG_EXECUTE_IF("CloudStorageEngine._submit_cumulative_compaction_task.sleep", { sleep(5); }) // Idempotent cleanup: remove task from tracker CompactionTaskTracker::instance()->remove_task(compaction_id); - std::lock_guard lock(_cumu_compaction_delay_mtx); - _cumu_compaction_thread_pool_used_threads--; - if (!is_large_task) { - _cumu_compaction_thread_pool_small_tasks_running--; + if (cumu_thread_counted) { + std::lock_guard lock(_cumu_compaction_delay_mtx); + _cumu_compaction_thread_pool_used_threads--; + if (!is_large_task) { + _cumu_compaction_thread_pool_small_tasks_running--; + } } g_cumu_compaction_running_task_count << -1; erase_submitted_cumu_compaction(); @@ -1050,6 +1053,7 @@ Status CloudStorageEngine::_submit_cumulative_compaction_task(const CloudTabletS do { std::lock_guard lock(_cumu_compaction_delay_mtx); _cumu_compaction_thread_pool_used_threads++; + cumu_thread_counted = true; if (config::large_cumu_compaction_task_min_thread_num > 1 && _cumu_compaction_thread_pool->max_threads() >= config::large_cumu_compaction_task_min_thread_num) { diff --git a/be/test/cloud/cloud_compaction_test.cpp b/be/test/cloud/cloud_compaction_test.cpp index 6c83d12f254..7db07e481ba 100644 --- a/be/test/cloud/cloud_compaction_test.cpp +++ b/be/test/cloud/cloud_compaction_test.cpp @@ -21,6 +21,7 @@ #include <gtest/gtest-test-part.h> #include <gtest/gtest.h> +#include <chrono> #include <memory> #include <mutex> #include <unordered_map> @@ -41,6 +42,7 @@ #include "storage/storage_policy.h" #include "storage/tablet/tablet_meta.h" #include "util/defer_op.h" +#include "util/threadpool.h" #include "util/time.h" #include "util/uid_util.h" @@ -224,6 +226,57 @@ static RowsetSharedPtr create_rowset(Version version, int num_segments, bool ove return rowset; } +TEST_F(CloudCompactionTest, cumulative_global_lock_failure_keeps_thread_count_balanced) { + ASSERT_TRUE(ThreadPoolBuilder("CumuCompactionTaskThreadPoolTest") + .set_min_threads(1) + .set_max_threads(1) + .build(&_engine._cumu_compaction_thread_pool) + .ok()); + + auto tablet_meta = std::make_shared<TabletMeta>(*_tablet_meta); + tablet_meta->_tablet_id = 12000; + auto tablet = std::make_shared<CloudTablet>(_engine, tablet_meta); + std::vector<RowsetSharedPtr> rowsets; + for (int64_t version = 0; version < 6; ++version) { + auto rowset = create_rowset(Version(version, version), 1, false, 41); + ASSERT_NE(rowset, nullptr); + rowsets.push_back(std::move(rowset)); + } + { + std::unique_lock lock(tablet->get_header_lock()); + tablet->add_rowsets(rowsets, false, lock, false); + } + tablet->set_cumulative_layer_point(0); + tablet->_approximate_num_rowsets = rowsets.size(); + tablet->_approximate_cumu_num_rowsets = rowsets.size(); + tablet->_approximate_cumu_num_deltas = rowsets.size(); + tablet->last_sync_time_s = 1; + + auto* sync_point = SyncPoint::get_instance(); + sync_point->enable_processing(); + sync_point->set_call_back("CloudMetaMgr::prepare_tablet_job", [](auto&& outcome) { + auto* response = try_any_cast<cloud::StartTabletJobResponse*>(outcome[1]); + response->mutable_status()->set_code(cloud::JOB_TABLET_BUSY); + response->mutable_status()->set_msg("injected global lock failure"); + auto* result = try_any_cast_ret<Status>(outcome); + result->first = Status::InternalError("injected global lock failure"); + result->second = true; + }); + Defer clear_sync_point {[&] { + sync_point->clear_all_call_backs(); + sync_point->disable_processing(); + }}; + + ASSERT_EQ(_engine._cumu_compaction_thread_pool_used_threads, 0); + ASSERT_EQ(_engine.submit_compaction_task(tablet, CompactionType::CUMULATIVE_COMPACTION), + Status::OK()); + ASSERT_TRUE(_engine._cumu_compaction_thread_pool->wait_for(std::chrono::seconds(5))); + + EXPECT_EQ(_engine._cumu_compaction_thread_pool_used_threads, 0); + EXPECT_EQ(_engine._cumu_compaction_thread_pool_small_tasks_running, 0); + EXPECT_FALSE(_engine.has_cumu_compaction(tablet->tablet_id())); +} + static RowsetSharedPtr create_delete_rowset(Version version) { auto rowset = create_rowset(version, 0, false, 0); DORIS_CHECK(rowset != nullptr); diff --git a/regression-test/suites/fault_injection_p0/cloud/test_cloud_cumu_compaction_global_lock_thread_count.groovy b/regression-test/suites/fault_injection_p0/cloud/test_cloud_cumu_compaction_global_lock_thread_count.groovy new file mode 100644 index 00000000000..40ce148850f --- /dev/null +++ b/regression-test/suites/fault_injection_p0/cloud/test_cloud_cumu_compaction_global_lock_thread_count.groovy @@ -0,0 +1,186 @@ +// 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. + +import groovy.json.JsonSlurper +import org.apache.doris.regression.suite.ClusterOptions +import org.apache.doris.regression.util.DebugPoint +import org.apache.doris.regression.util.NodeType + +suite('test_cloud_cumu_compaction_global_lock_thread_count', 'docker') { + def options = new ClusterOptions() + options.cloudMode = true + options.enableDebugPoints() + options.beConfigs += [ + 'enable_java_support=false', + 'cumulative_compaction_min_deltas=2', + 'cumulative_compaction_max_deltas=3', + 'max_cumu_compaction_threads=2', + 'large_cumu_compaction_task_min_thread_num=2', + 'large_cumu_compaction_task_row_num_threshold=1', + 'disable_auto_compaction=true', + ] + options.beNum = 3 + + docker(options) { + GetDebugPoint().clearDebugPointsForAllBEs() + def backends = sql_return_maparray('SHOW BACKENDS') + def lockHolderBe = backends[0] + def lockFailedBe = backends[1] + assertNotNull(lockHolderBe) + assertNotNull(lockFailedBe) + + def getCompactionStatus = { be, tabletId -> + def (code, out, err) = be_get_compaction_status(be.Host, be.HttpPort, tabletId) + assertEquals(0, code, "failed to get compaction status: ${out}, ${err}") + return new JsonSlurper().parseText(out.trim()) + } + + def waitForCompactionRunning = { be, tabletId -> + def deadline = System.currentTimeMillis() + 30000 + while (System.currentTimeMillis() < deadline) { + if (getCompactionStatus(be, tabletId).run_status) { + return + } + sleep(100) + } + assertTrue(false, "compaction did not start for tablet ${tabletId}") + } + + def waitForCompactionFinished = { be, tabletId -> + def deadline = System.currentTimeMillis() + 30000 + while (System.currentTimeMillis() < deadline) { + if (!getCompactionStatus(be, tabletId).run_status) { + return + } + sleep(100) + } + assertTrue(false, "compaction did not finish for tablet ${tabletId}") + } + + def getTabletStatus = { be, tabletId -> + def (code, out, err) = be_show_tablet_status(be.Host, be.HttpPort, tabletId) + assertEquals(0, code, "failed to get tablet status: ${out}, ${err}") + return new JsonSlurper().parseText(out.trim()) + } + + def getRowsetCount = { be, tabletId -> + def status = getTabletStatus(be, tabletId) + assertTrue(status.rowsets instanceof List) + return status.rowsets.size() + } + + def createTable = { table -> + sql "DROP TABLE IF EXISTS ${table} FORCE" + sql """ + CREATE TABLE ${table} ( + k INT, + v INT + ) DUPLICATE KEY(k) + DISTRIBUTED BY HASH(k) BUCKETS 1 + PROPERTIES ( + "replication_num" = "1", + "disable_auto_compaction" = "true" + ) + """ + for (int value = 0; value < 5; value++) { + sql "INSERT INTO ${table} VALUES (${value}, ${value})" + } + sql "SELECT COUNT(*) FROM ${table}" + def tablet = sql_return_maparray("SHOW TABLETS FROM ${table}")[0] + return tablet.TabletId + } + + def runCumulativeCompaction = { be, tabletId -> + def (code, out, err) = be_run_cumulative_compaction(be.Host, be.HttpPort, tabletId) + assertEquals(0, code, "failed to submit cumulative compaction: ${out}, ${err}") + } + + def conflictTabletId = createTable('test_cloud_cumu_compaction_global_lock_conflict') + def conflictRowsetsBefore = getRowsetCount(lockFailedBe, conflictTabletId) + def holderRowsetsBefore = getRowsetCount(lockHolderBe, conflictTabletId) + def blockModifyRowsets = 'CloudCumulativeCompaction::modify_rowsets.enable_spin_wait' + def blockModifyRowsetsSwitch = 'CloudCumulativeCompaction::modify_rowsets.block' + def holdTaskAfterExecution = 'CloudStorageEngine._submit_cumulative_compaction_task.sleep' + + try { + DebugPoint.enableDebugPoint(lockHolderBe.Host, lockHolderBe.HttpPort.toInteger(), + NodeType.BE, blockModifyRowsets) + DebugPoint.enableDebugPoint(lockHolderBe.Host, lockHolderBe.HttpPort.toInteger(), + NodeType.BE, blockModifyRowsetsSwitch) + + runCumulativeCompaction(lockHolderBe, conflictTabletId) + waitForCompactionRunning(lockHolderBe, conflictTabletId) + + runCumulativeCompaction(lockFailedBe, conflictTabletId) + sleep(1000) + waitForCompactionFinished(lockFailedBe, conflictTabletId) + assertEquals(conflictRowsetsBefore, getRowsetCount(lockFailedBe, conflictTabletId)) + + DebugPoint.disableDebugPoint(lockHolderBe.Host, lockHolderBe.HttpPort.toInteger(), + NodeType.BE, blockModifyRowsetsSwitch) + DebugPoint.disableDebugPoint(lockHolderBe.Host, lockHolderBe.HttpPort.toInteger(), + NodeType.BE, blockModifyRowsets) + waitForCompactionFinished(lockHolderBe, conflictTabletId) + assertTrue(getRowsetCount(lockHolderBe, conflictTabletId) < holderRowsetsBefore) + + def runningTabletId = createTable('test_cloud_cumu_compaction_global_lock_thread_holder') + def candidateTabletId = createTable('test_cloud_cumu_compaction_global_lock_thread_candidate') + def candidateRowsetsBefore = getRowsetCount(lockFailedBe, candidateTabletId) + def runningRowsetsBefore = getRowsetCount(lockFailedBe, runningTabletId) + + DebugPoint.enableDebugPoint(lockFailedBe.Host, lockFailedBe.HttpPort.toInteger(), + NodeType.BE, holdTaskAfterExecution) + DebugPoint.enableDebugPoint(lockFailedBe.Host, lockFailedBe.HttpPort.toInteger(), + NodeType.BE, blockModifyRowsets) + DebugPoint.enableDebugPoint(lockFailedBe.Host, lockFailedBe.HttpPort.toInteger(), + NodeType.BE, blockModifyRowsetsSwitch) + + runCumulativeCompaction(lockFailedBe, runningTabletId) + waitForCompactionRunning(lockFailedBe, runningTabletId) + + DebugPoint.disableDebugPoint(lockFailedBe.Host, lockFailedBe.HttpPort.toInteger(), + NodeType.BE, blockModifyRowsetsSwitch) + DebugPoint.disableDebugPoint(lockFailedBe.Host, lockFailedBe.HttpPort.toInteger(), + NodeType.BE, blockModifyRowsets) + def deadline = System.currentTimeMillis() + 30000 + while (System.currentTimeMillis() < deadline && + getRowsetCount(lockFailedBe, runningTabletId) >= runningRowsetsBefore) { + sleep(100) + } + assertTrue(getRowsetCount(lockFailedBe, runningTabletId) < runningRowsetsBefore) + assertTrue(getCompactionStatus(lockFailedBe, runningTabletId).run_status) + + runCumulativeCompaction(lockFailedBe, candidateTabletId) + sleep(1000) + assertEquals(candidateRowsetsBefore, getRowsetCount(lockFailedBe, candidateTabletId), + 'a second large compaction must be delayed while the first task is active') + waitForCompactionFinished(lockFailedBe, candidateTabletId) + waitForCompactionFinished(lockFailedBe, runningTabletId) + } finally { + DebugPoint.disableDebugPoint(lockHolderBe.Host, lockHolderBe.HttpPort.toInteger(), + NodeType.BE, blockModifyRowsetsSwitch) + DebugPoint.disableDebugPoint(lockHolderBe.Host, lockHolderBe.HttpPort.toInteger(), + NodeType.BE, blockModifyRowsets) + DebugPoint.disableDebugPoint(lockFailedBe.Host, lockFailedBe.HttpPort.toInteger(), + NodeType.BE, holdTaskAfterExecution) + DebugPoint.disableDebugPoint(lockFailedBe.Host, lockFailedBe.HttpPort.toInteger(), + NodeType.BE, blockModifyRowsetsSwitch) + DebugPoint.disableDebugPoint(lockFailedBe.Host, lockFailedBe.HttpPort.toInteger(), + NodeType.BE, blockModifyRowsets) + } + } +} --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
