From 728eb47b99646d84661e943ad99bb3391b2d2f90 Mon Sep 17 00:00:00 2001 From: Yukang-Lian Date: Fri, 17 Apr 2026 19:42:08 +0800 Subject: [PATCH 1/3] [feat](compaction) Support ADMIN COMPACT TABLE type='FULL' and enable it in cloud mode ADMIN COMPACT TABLE previously only accepted BASE and CUMULATIVE in local mode and threw Unsupported operation in cloud mode, forcing users to curl each BE's HTTP API to run manual compaction. This change unifies the FE SQL entry so that base/cumulative/full all work on both local and cloud. FE - AdminCompactTableCommand: add FULL to CompactionType, relax analyzeWhere, map FULL->"full" in getCompactionType, drop the cloud-mode gate so the command dispatches via the normal path. - CloudEnv.compactTable: override Env.compactTable to walk partitions/ indices/tablets, resolve the primary BE via Replica.getBackendId, and raise a DdlException if nothing was dispatched (e.g. the current compute group has no available BE). BE - submit_table_compaction_callback: strictly match base/cumulative/full and reject unknown types instead of silently downgrading non-"base" to cumulative. Full goes through submit_compaction_task(force=false, eager=true, trigger_method=MANUAL) after setting last_full_compaction_schedule_time, matching the HTTP API default. - cloud_submit_table_compaction_callback: new callback mirroring the local one but calling CloudStorageEngine::submit_compaction_task; honors the cloud HTTP convention that base/cumu need sync_delete_bitmap while full does not. - agent_server::cloud_start_workers: register the previously stubbed-out SUBMIT_TABLE_COMPACTION worker (drops the plat1ko TODO). Thrift is untouched: TCompactionReq.type is already optional string, so the existing "base"/"cumulative"/"full" wire format remains compatible. This implies a release order of "all BEs upgraded first, then FE" - a new FE against an old BE would see "full" silently downgraded (local) or dropped (cloud). Tests - regression-test/suites/compaction/test_admin_compact_table.groovy: end-to-end coverage for base/cumulative/full under the local engine plus negative cases (unknown type, missing WHERE). - regression-test/suites/cloud_p0/compaction/test_cloud_admin_compact_table.groovy: same matrix under the cloud engine, with a SELECT before each poll to force BE to sync rowsets from Meta Service. --- be/src/agent/agent_server.cpp | 5 +- be/src/agent/task_worker_pool.cpp | 109 ++++++++++-- be/src/agent/task_worker_pool.h | 3 + .../apache/doris/cloud/catalog/CloudEnv.java | 63 +++++++ .../commands/AdminCompactTableCommand.java | 39 ++--- .../test_cloud_admin_compact_table.out | 14 ++ .../compaction/test_admin_compact_table.out | 18 ++ .../test_cloud_admin_compact_table.groovy | 158 +++++++++++++++++ .../test_admin_compact_table.groovy | 163 ++++++++++++++++++ 9 files changed, 537 insertions(+), 35 deletions(-) create mode 100644 regression-test/data/cloud_p0/compaction/test_cloud_admin_compact_table.out create mode 100644 regression-test/data/compaction/test_admin_compact_table.out create mode 100644 regression-test/suites/cloud_p0/compaction/test_cloud_admin_compact_table.groovy create mode 100644 regression-test/suites/compaction/test_admin_compact_table.groovy diff --git a/be/src/agent/agent_server.cpp b/be/src/agent/agent_server.cpp index a036adc013f62f..bfc8c9d278294e 100644 --- a/be/src/agent/agent_server.cpp +++ b/be/src/agent/agent_server.cpp @@ -207,7 +207,10 @@ void AgentServer::cloud_start_workers(CloudStorageEngine& engine, ExecEnv* exec_ _workers[TTaskType::PUSH] = std::make_unique( "PUSH", config::delete_worker_count, [&engine](auto&& task) { cloud_push_callback(engine, task); }); - // TODO(plat1ko): SUBMIT_TABLE_COMPACTION worker + + _workers[TTaskType::COMPACTION] = std::make_unique( + "SUBMIT_TABLE_COMPACTION", 1, + [&engine](auto&& task) { cloud_submit_table_compaction_callback(engine, task); }); _workers[TTaskType::ALTER] = std::make_unique( "ALTER_TABLE", config::alter_tablet_worker_count, diff --git a/be/src/agent/task_worker_pool.cpp b/be/src/agent/task_worker_pool.cpp index 45720833eab8c9..bb974927ddec9c 100644 --- a/be/src/agent/task_worker_pool.cpp +++ b/be/src/agent/task_worker_pool.cpp @@ -1605,28 +1605,115 @@ void submit_table_compaction_callback(StorageEngine& engine, const TAgentTaskReq const auto& compaction_req = req.compaction_req; LOG(INFO) << "get compaction task. signature=" << req.signature - << ", compaction_type=" << compaction_req.type; + << ", compaction_type=" << compaction_req.type + << ", tablet_id=" << compaction_req.tablet_id; CompactionType compaction_type; if (compaction_req.type == "base") { compaction_type = CompactionType::BASE_COMPACTION; - } else { + } else if (compaction_req.type == "cumulative") { compaction_type = CompactionType::CUMULATIVE_COMPACTION; + } else if (compaction_req.type == "full") { + compaction_type = CompactionType::FULL_COMPACTION; + } else { + LOG(WARNING) << "unknown compaction type: " << compaction_req.type + << ", tablet_id=" << compaction_req.tablet_id; + return; } auto tablet_ptr = engine.tablet_manager()->get_tablet(compaction_req.tablet_id); - if (tablet_ptr != nullptr) { - auto* data_dir = tablet_ptr->data_dir(); - if (!tablet_ptr->can_do_compaction(data_dir->path_hash(), compaction_type)) { - LOG(WARNING) << "could not do compaction. tablet_id=" << tablet_ptr->tablet_id() - << ", compaction_type=" << compaction_type; - return; - } + if (tablet_ptr == nullptr) { + LOG(WARNING) << "tablet not found. tablet_id=" << compaction_req.tablet_id; + return; + } - Status status = engine.submit_compaction_task(tablet_ptr, compaction_type, false); + if (compaction_type == CompactionType::FULL_COMPACTION) { + // Full compaction goes through the dedicated threadpool path (align with + // compaction_action.cpp _handle_run_compaction). `force=false` keeps the + // admission under permit limiter, matching the HTTP API default. + tablet_ptr->set_last_full_compaction_schedule_time(UnixMillis()); + Status status = engine.submit_compaction_task( + tablet_ptr, CompactionType::FULL_COMPACTION, + /*force=*/false, /*eager=*/true, /*trigger_method=*/1); if (!status.ok()) { - LOG(WARNING) << "failed to submit table compaction task. error=" << status; + LOG(WARNING) << "failed to submit full compaction task. tablet_id=" + << tablet_ptr->tablet_id() << ", error=" << status; } + return; + } + + // base / cumulative + auto* data_dir = tablet_ptr->data_dir(); + if (!tablet_ptr->can_do_compaction(data_dir->path_hash(), compaction_type)) { + LOG(WARNING) << "could not do compaction. tablet_id=" << tablet_ptr->tablet_id() + << ", compaction_type=" << compaction_type; + return; + } + + Status status = engine.submit_compaction_task(tablet_ptr, compaction_type, false); + if (!status.ok()) { + LOG(WARNING) << "failed to submit table compaction task. error=" << status; + } +} + +void cloud_submit_table_compaction_callback(CloudStorageEngine& engine, + const TAgentTaskRequest& req) { + const auto& compaction_req = req.compaction_req; + + LOG(INFO) << "get cloud compaction task. signature=" << req.signature + << ", compaction_type=" << compaction_req.type + << ", tablet_id=" << compaction_req.tablet_id; + + CompactionType compaction_type; + if (compaction_req.type == "base") { + compaction_type = CompactionType::BASE_COMPACTION; + } else if (compaction_req.type == "cumulative") { + compaction_type = CompactionType::CUMULATIVE_COMPACTION; + } else if (compaction_req.type == "full") { + compaction_type = CompactionType::FULL_COMPACTION; + } else { + LOG(WARNING) << "unknown cloud compaction type: " << compaction_req.type + << ", tablet_id=" << compaction_req.tablet_id; + return; + } + + // Mirror cloud_compaction_action::_handle_run_compaction: base/cumu needs the + // delete bitmap synced eagerly, full does not (FullCompaction re-syncs itself). + bool sync_delete_bitmap = compaction_type != CompactionType::FULL_COMPACTION; + auto tablet_res = engine.tablet_mgr().get_tablet(compaction_req.tablet_id, + /*warmup_data=*/false, + sync_delete_bitmap); + if (!tablet_res.has_value()) { + LOG(WARNING) << "failed to get cloud tablet. tablet_id=" << compaction_req.tablet_id + << ", error=" << tablet_res.error(); + return; + } + CloudTabletSPtr tablet = std::move(tablet_res).value(); + if (tablet == nullptr) { + LOG(WARNING) << "cloud tablet not found. tablet_id=" << compaction_req.tablet_id; + return; + } + + switch (compaction_type) { + case CompactionType::BASE_COMPACTION: + tablet->set_last_base_compaction_schedule_time(UnixMillis()); + break; + case CompactionType::CUMULATIVE_COMPACTION: + tablet->set_last_cumu_compaction_schedule_time(UnixMillis()); + break; + case CompactionType::FULL_COMPACTION: + tablet->set_last_full_compaction_schedule_time(UnixMillis()); + break; + default: + break; + } + + Status status = engine.submit_compaction_task(tablet, compaction_type, + /*trigger_method=*/1); + if (!status.ok()) { + LOG(WARNING) << "failed to submit cloud compaction task. tablet_id=" + << tablet->tablet_id() << ", type=" << compaction_req.type + << ", error=" << status; } } diff --git a/be/src/agent/task_worker_pool.h b/be/src/agent/task_worker_pool.h index 4f8787c1bd6c19..a389450dd55bd5 100644 --- a/be/src/agent/task_worker_pool.h +++ b/be/src/agent/task_worker_pool.h @@ -158,6 +158,9 @@ void move_dir_callback(CloudStorageEngine& engine, ExecEnv* env, const TAgentTas void submit_table_compaction_callback(StorageEngine& engine, const TAgentTaskRequest& req); +void cloud_submit_table_compaction_callback(CloudStorageEngine& engine, + const TAgentTaskRequest& req); + void push_storage_policy_callback(StorageEngine& engine, const TAgentTaskRequest& req); void push_index_policy_callback(const TAgentTaskRequest& req); diff --git a/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudEnv.java b/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudEnv.java index 07d4fe552d674c..69e050b9c298f3 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudEnv.java +++ b/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudEnv.java @@ -18,8 +18,15 @@ package org.apache.doris.cloud.catalog; import org.apache.doris.analysis.ResourceTypeEnum; +import org.apache.doris.catalog.Database; import org.apache.doris.catalog.Env; import org.apache.doris.catalog.EnvFactory; +import org.apache.doris.catalog.MaterializedIndex; +import org.apache.doris.catalog.MaterializedIndex.IndexExtState; +import org.apache.doris.catalog.OlapTable; +import org.apache.doris.catalog.Partition; +import org.apache.doris.catalog.Replica; +import org.apache.doris.catalog.Tablet; import org.apache.doris.cloud.CacheHotspotManager; import org.apache.doris.cloud.CloudWarmUpJob; import org.apache.doris.cloud.CloudWarmUpJob.JobState; @@ -45,6 +52,9 @@ import org.apache.doris.qe.ConnectContext; import org.apache.doris.system.Frontend; import org.apache.doris.system.SystemInfoService.HostInfo; +import org.apache.doris.task.AgentBatchTask; +import org.apache.doris.task.AgentTaskExecutor; +import org.apache.doris.task.CompactionTask; import com.google.common.base.Preconditions; import com.google.common.base.Strings; @@ -489,4 +499,57 @@ protected void cloneClusterSnapshot() throws Exception { this.cloudSnapshotHandler.cloneSnapshot(this.clusterSnapshotFile); } } + + @Override + public void compactTable(String dbName, String tableName, String type, List partitionNames) + throws DdlException { + Database db = getInternalCatalog().getDbOrDdlException(dbName); + OlapTable olapTable = db.getOlapTableOrDdlException(tableName); + + AgentBatchTask batchTask = new AgentBatchTask(); + int dispatchedCount = 0; + olapTable.readLock(); + try { + LOG.info("Cloud table compaction. db={}, table={}, partitions={}, type={}", + dbName, tableName, partitionNames, type); + for (String parName : partitionNames) { + Partition partition = olapTable.getPartition(parName); + if (partition == null) { + throw new DdlException("partition[" + parName + "] not exist in table[" + tableName + "]"); + } + + for (MaterializedIndex idx : partition.getMaterializedIndices(IndexExtState.VISIBLE)) { + for (Tablet tablet : idx.getTablets()) { + // Cloud: each tablet has only one CloudReplica (primary BE). The backend + // resolution depends on the current compute group (ConnectContext); skip + // the tablet rather than aborting so other tablets can still dispatch. + for (Replica replica : tablet.getReplicas()) { + long beId; + try { + beId = replica.getBackendId(); + } catch (UserException e) { + LOG.warn("skip tablet {} for cloud compaction, no available BE: {}", + tablet.getId(), e.getMessage()); + continue; + } + CompactionTask compactionTask = new CompactionTask( + beId, db.getId(), olapTable.getId(), partition.getId(), + idx.getId(), tablet.getId(), + olapTable.getSchemaHashByIndexId(idx.getId()), type); + batchTask.addTask(compactionTask); + dispatchedCount++; + } + } + } + } + } finally { + olapTable.readUnlock(); + } + + if (dispatchedCount == 0) { + throw new DdlException("no tablet dispatched for compaction; " + + "the current compute group may have no available BE"); + } + AgentTaskExecutor.submit(batchTask); + } } diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AdminCompactTableCommand.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AdminCompactTableCommand.java index f41d24f4b6278b..0ca208575f10e6 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AdminCompactTableCommand.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AdminCompactTableCommand.java @@ -19,7 +19,6 @@ import org.apache.doris.catalog.Env; import org.apache.doris.common.AnalysisException; -import org.apache.doris.common.DdlException; import org.apache.doris.common.ErrorCode; import org.apache.doris.common.ErrorReport; import org.apache.doris.common.UserException; @@ -33,16 +32,12 @@ import org.apache.doris.qe.ConnectContext; import org.apache.doris.qe.StmtExecutor; -import org.apache.logging.log4j.LogManager; -import org.apache.logging.log4j.Logger; - import java.util.List; /** * AdminCompactTableCommand */ public class AdminCompactTableCommand extends Command implements ForwardWithSync { - private static final Logger LOG = LogManager.getLogger(AdminCompactTableCommand.class); private TableRefInfo tableRefInfo; private EqualTo where; @@ -51,7 +46,8 @@ public class AdminCompactTableCommand extends Command implements ForwardWithSync */ public enum CompactionType { CUMULATIVE, - BASE + BASE, + FULL } private CompactionType typeFilter; @@ -92,12 +88,12 @@ private void validate(ConnectContext ctx) throws UserException { // analyze where clause if not null if (where == null) { throw new AnalysisException("Compaction type must be specified in" - + " Where clause like: type = 'BASE/CUMULATIVE'"); + + " Where clause like: type = 'BASE/CUMULATIVE/FULL'"); } if (!analyzeWhere()) { throw new AnalysisException( - "Where clause should looks like: type = 'BASE/CUMULATIVE'"); + "Where clause should looks like: type = 'BASE/CUMULATIVE/FULL'"); } } @@ -108,11 +104,9 @@ private boolean analyzeWhere() { return false; } - if (typeFilter == null || (typeFilter != CompactionType.CUMULATIVE && typeFilter != CompactionType.BASE)) { - return false; - } - - return true; + return typeFilter == CompactionType.CUMULATIVE + || typeFilter == CompactionType.BASE + || typeFilter == CompactionType.FULL; } @Override @@ -121,16 +115,15 @@ public R accept(PlanVisitor visitor, C context) { } private String getCompactionType() { - if (typeFilter == CompactionType.CUMULATIVE) { - return "cumulative"; - } else { - return "base"; + switch (typeFilter) { + case CUMULATIVE: + return "cumulative"; + case BASE: + return "base"; + case FULL: + return "full"; + default: + throw new IllegalStateException("unexpected compaction type: " + typeFilter); } } - - @Override - protected void checkSupportedInCloudMode(ConnectContext ctx) throws DdlException { - LOG.info("AdminCompactTableCommand not supported in cloud mode"); - throw new DdlException("Unsupported operation"); - } } diff --git a/regression-test/data/cloud_p0/compaction/test_cloud_admin_compact_table.out b/regression-test/data/cloud_p0/compaction/test_cloud_admin_compact_table.out new file mode 100644 index 00000000000000..3403a3055a2b61 --- /dev/null +++ b/regression-test/data/cloud_p0/compaction/test_cloud_admin_compact_table.out @@ -0,0 +1,14 @@ +-- This file is automatically generated. You should know what you did if you want to edit this +-- !select_count -- +8 + +-- !select_all -- +1 1 +2 2 +3 3 +4 4 +5 5 +6 6 +7 7 +8 8 + diff --git a/regression-test/data/compaction/test_admin_compact_table.out b/regression-test/data/compaction/test_admin_compact_table.out new file mode 100644 index 00000000000000..c6e3430953e3fc --- /dev/null +++ b/regression-test/data/compaction/test_admin_compact_table.out @@ -0,0 +1,18 @@ +-- This file is automatically generated. You should know what you did if you want to edit this +-- !select_count -- +12 + +-- !select_all -- +1 1 +2 2 +3 3 +4 4 +5 5 +6 6 +7 7 +8 8 +9 9 +10 10 +11 11 +12 12 + diff --git a/regression-test/suites/cloud_p0/compaction/test_cloud_admin_compact_table.groovy b/regression-test/suites/cloud_p0/compaction/test_cloud_admin_compact_table.groovy new file mode 100644 index 00000000000000..37cda47886cadd --- /dev/null +++ b/regression-test/suites/cloud_p0/compaction/test_cloud_admin_compact_table.groovy @@ -0,0 +1,158 @@ +// 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. + +// End-to-end test for `ADMIN COMPACT TABLE ... WHERE type = 'BASE'|'CUMULATIVE'|'FULL'` +// under the cloud (storage-compute split) engine. +// +// Mirrors test_admin_compact_table.groovy but accounts for cloud specifics: +// - each tablet has a single CloudReplica; FE routes via CloudReplica.getBackendId() +// - the BE does not hold the full rowset list until something forces a sync, +// so we issue a SELECT before polling +suite("test_cloud_admin_compact_table", "p0") { + if (!isCloudMode()) { + logger.info("not cloud mode, skip this test") + return + } + + def tableName = "test_cloud_admin_compact_table" + def partName = tableName + + sql "DROP TABLE IF EXISTS ${tableName}" + sql """ + CREATE TABLE ${tableName} ( + k INT, + v INT + ) DUPLICATE KEY(k) + DISTRIBUTED BY HASH(k) BUCKETS 1 + PROPERTIES ( + "replication_num" = "1", + "disable_auto_compaction" = "true" + ) + """ + + def tablets = sql_return_maparray "SHOW TABLETS FROM ${tableName}" + assertEquals(1, tablets.size()) + def tabletId = tablets[0].TabletId + def backendId = tablets[0].BackendId + + def backendId_to_backendIP = [:] + def backendId_to_backendHttpPort = [:] + getBackendIpHttpPort(backendId_to_backendIP, backendId_to_backendHttpPort) + def beHost = backendId_to_backendIP["${backendId}"] + def bePort = backendId_to_backendHttpPort["${backendId}"] + + def showTabletCompaction = { + // Force a read so BE syncs rowsets from MS before we inspect them. + sql "SELECT COUNT(*) FROM ${tableName}" + def (code, stdout, stderr) = be_show_tablet_status(beHost, bePort, tabletId) + assertEquals(0, code) + return parseJson(stdout.trim()) + } + + def countDataRowsets = { json -> + return json.rowsets.findAll { it.contains(" DATA ") }.size() + } + + def EPOCH_TIME = "1970-01-01 08:00:00.000" + + // Poll until BOTH `last__success_time` is updated AND the rowset count + // has decreased. Cloud rowset list refreshes after MS confirms the merge, so + // requiring both prevents false positives when only the schedule time ticks. + def waitCompaction = { String type, int beforeCount, int timeoutSec = 90 -> + def deadline = System.currentTimeMillis() + timeoutSec * 1000L + def last + while (System.currentTimeMillis() < deadline) { + last = showTabletCompaction() + def success = last["last ${type} success time"] + def timeUpdated = success != null && success != EPOCH_TIME + def merged = countDataRowsets(last) < beforeCount + if (timeUpdated && merged) { + return last + } + sleep(1000) + } + return last + } + + // ---------- load data ---------- + for (int i = 1; i <= 5; i++) { + sql "INSERT INTO ${tableName} VALUES (${i}, ${i})" + } + + def beforeCumu = showTabletCompaction() + def rowsetsBeforeCumu = countDataRowsets(beforeCumu) + assertTrue(rowsetsBeforeCumu >= 5, + "expected >= 5 data rowsets before cumulative, got ${rowsetsBeforeCumu}") + + // ---------- case 1: cumulative ---------- + sql "ADMIN COMPACT TABLE ${tableName} PARTITION (${partName}) WHERE type = 'cumulative'" + def afterCumu = waitCompaction("cumulative", rowsetsBeforeCumu) + assertNotEquals(EPOCH_TIME, afterCumu["last cumulative success time"]) + assertTrue(countDataRowsets(afterCumu) < rowsetsBeforeCumu, + "cumulative did not reduce rowset count: ${afterCumu.rowsets}") + + // ---------- case 2: full ---------- + for (int i = 6; i <= 8; i++) { + sql "INSERT INTO ${tableName} VALUES (${i}, ${i})" + } + def beforeFull = showTabletCompaction() + def rowsetsBeforeFull = countDataRowsets(beforeFull) + assertTrue(rowsetsBeforeFull >= 2, + "expected >= 2 data rowsets before full, got ${rowsetsBeforeFull}") + + sql "ADMIN COMPACT TABLE ${tableName} PARTITION (${partName}) WHERE type = 'full'" + def afterFull = waitCompaction("full", rowsetsBeforeFull) + assertNotEquals(EPOCH_TIME, afterFull["last full success time"]) + assertEquals("[OK]", afterFull["last full status"]) + assertTrue(countDataRowsets(afterFull) < rowsetsBeforeFull, + "cloud full did not reduce rowset count: ${afterFull.rowsets}") + + // ---------- case 3: base ---------- + sql "ADMIN COMPACT TABLE ${tableName} PARTITION (${partName}) WHERE type = 'base'" + def afterBase = null + for (int i = 0; i < 30; i++) { + afterBase = showTabletCompaction() + if (afterBase["last base status"] != "" || + afterBase["last base success time"] != EPOCH_TIME || + afterBase["last base failure time"] != EPOCH_TIME) { + break + } + sleep(1000) + } + // Cloud base may legitimately be rejected for insufficient input rowsets + // (e.g. `[E-808]insufficent compaction input rowset`). We only verify the + // FE->BE path dispatched the task. + def baseStatus = afterBase["last base status"] + assertTrue(baseStatus == "[OK]" || baseStatus.startsWith("[E-"), + "unexpected base status in cloud: ${baseStatus}") + + // ---------- data correctness ---------- + qt_select_count "SELECT COUNT(*) FROM ${tableName}" + qt_select_all "SELECT * FROM ${tableName} ORDER BY k" + + // ---------- negative: unknown type ---------- + test { + sql "ADMIN COMPACT TABLE ${tableName} PARTITION (${partName}) WHERE type = 'UNKNOWN'" + exception "BASE/CUMULATIVE/FULL" + } + + // ---------- negative: no WHERE clause ---------- + test { + sql "ADMIN COMPACT TABLE ${tableName} PARTITION (${partName})" + exception "type" + } +} diff --git a/regression-test/suites/compaction/test_admin_compact_table.groovy b/regression-test/suites/compaction/test_admin_compact_table.groovy new file mode 100644 index 00000000000000..2d6670c4123a9a --- /dev/null +++ b/regression-test/suites/compaction/test_admin_compact_table.groovy @@ -0,0 +1,163 @@ +// 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. + +// End-to-end test for `ADMIN COMPACT TABLE ... WHERE type = 'BASE'|'CUMULATIVE'|'FULL'` +// under the non-cloud (local storage) engine. +// +// Covers: +// - all three compaction types can be triggered from the FE SQL entry +// - rowsets are actually merged (not just last_*_success_time updated) +// - data correctness after compaction +// - negative cases: unknown type, multi-partition, missing WHERE +suite("test_admin_compact_table", "p0") { + if (isCloudMode()) { + return + } + + def tableName = "test_admin_compact_table" + def partName = tableName + + sql "DROP TABLE IF EXISTS ${tableName}" + sql """ + CREATE TABLE ${tableName} ( + k INT, + v INT + ) DUPLICATE KEY(k) + DISTRIBUTED BY HASH(k) BUCKETS 1 + PROPERTIES ( + "replication_num" = "1", + "disable_auto_compaction" = "true" + ) + """ + + def tablets = sql_return_maparray "SHOW TABLETS FROM ${tableName}" + assertEquals(1, tablets.size()) + def tabletId = tablets[0].TabletId + def backendId = tablets[0].BackendId + + def backendId_to_backendIP = [:] + def backendId_to_backendHttpPort = [:] + getBackendIpHttpPort(backendId_to_backendIP, backendId_to_backendHttpPort) + def beHost = backendId_to_backendIP["${backendId}"] + def bePort = backendId_to_backendHttpPort["${backendId}"] + + // Pull compaction show JSON. Fields of interest: + // rowsets: array of "[vL-vR] ... DATA ..." strings + // last__success_time / last__status + def showTabletCompaction = { + def (code, stdout, stderr) = be_show_tablet_status(beHost, bePort, tabletId) + assertEquals(0, code) + return parseJson(stdout.trim()) + } + + def countDataRowsets = { json -> + return json.rowsets.findAll { it.contains(" DATA ") }.size() + } + + def EPOCH_TIME = "1970-01-01 08:00:00.000" + + // Poll until BOTH `last__success_time` is updated AND the rowset count + // has decreased (i.e. the compaction actually merged something). Success time + // alone can be updated before the new rowset is installed, so we require both + // to avoid race-y assertions downstream. + def waitCompaction = { String type, int beforeCount, int timeoutSec = 60 -> + def deadline = System.currentTimeMillis() + timeoutSec * 1000L + def last + while (System.currentTimeMillis() < deadline) { + last = showTabletCompaction() + def success = last["last ${type} success time"] + def timeUpdated = success != null && success != EPOCH_TIME + def merged = countDataRowsets(last) < beforeCount + if (timeUpdated && merged) { + return last + } + sleep(500) + } + return last + } + + // ---------- load data to produce multiple rowsets ---------- + // 8 inserts -> 8 additional rowsets on top of [0-1] (initial empty rowset) + for (int i = 1; i <= 8; i++) { + sql "INSERT INTO ${tableName} VALUES (${i}, ${i})" + } + + def beforeCumu = showTabletCompaction() + def rowsetsBeforeCumu = countDataRowsets(beforeCumu) + assertTrue(rowsetsBeforeCumu >= 8, + "expected >= 8 data rowsets before cumulative, got ${rowsetsBeforeCumu}") + assertEquals(EPOCH_TIME, beforeCumu["last cumulative success time"]) + + // ---------- case 1: cumulative ---------- + sql "ADMIN COMPACT TABLE ${tableName} PARTITION (${partName}) WHERE type = 'cumulative'" + def afterCumu = waitCompaction("cumulative", rowsetsBeforeCumu) + assertNotEquals(EPOCH_TIME, afterCumu["last cumulative success time"]) + assertEquals("[OK]", afterCumu["last cumulative status"]) + assertTrue(countDataRowsets(afterCumu) < rowsetsBeforeCumu, + "cumulative did not reduce rowset count: ${afterCumu.rowsets}") + + // ---------- case 2: full ---------- + // Add more rowsets after cumulative so full has work to do. + for (int i = 9; i <= 12; i++) { + sql "INSERT INTO ${tableName} VALUES (${i}, ${i})" + } + def beforeFull = showTabletCompaction() + def rowsetsBeforeFull = countDataRowsets(beforeFull) + assertTrue(rowsetsBeforeFull >= 2, + "expected >= 2 data rowsets before full, got ${rowsetsBeforeFull}") + + sql "ADMIN COMPACT TABLE ${tableName} PARTITION (${partName}) WHERE type = 'full'" + def afterFull = waitCompaction("full", rowsetsBeforeFull) + assertNotEquals(EPOCH_TIME, afterFull["last full success time"]) + assertEquals("[OK]", afterFull["last full status"]) + assertTrue(countDataRowsets(afterFull) < rowsetsBeforeFull, + "full did not reduce rowset count: ${afterFull.rowsets}") + + // ---------- case 3: base (may be a no-op depending on rowset shape) ---------- + sql "ADMIN COMPACT TABLE ${tableName} PARTITION (${partName}) WHERE type = 'base'" + // Poll a bit: base may be rejected (initial rowset empty) -> last base status non-empty. + def afterBase = null + for (int i = 0; i < 20; i++) { + afterBase = showTabletCompaction() + if (afterBase["last base status"] != "") { + break + } + sleep(500) + } + assertNotNull(afterBase, "base compaction never scheduled") + // Either success or a legitimate BE rejection; both prove the FE->BE path + // accepted the "base" type and dispatched it. + def baseStatus = afterBase["last base status"] + assertTrue(baseStatus == "[OK]" || baseStatus.startsWith("[E-"), + "unexpected base status: ${baseStatus}") + + // ---------- data correctness ---------- + qt_select_count "SELECT COUNT(*) FROM ${tableName}" + qt_select_all "SELECT * FROM ${tableName} ORDER BY k" + + // ---------- negative: unknown type ---------- + test { + sql "ADMIN COMPACT TABLE ${tableName} PARTITION (${partName}) WHERE type = 'UNKNOWN'" + exception "BASE/CUMULATIVE/FULL" + } + + // ---------- negative: no WHERE ---------- + test { + sql "ADMIN COMPACT TABLE ${tableName} PARTITION (${partName})" + exception "type" + } +} From 247e5ea4c7dfce671ccfc6c730dca58a67d73ce5 Mon Sep 17 00:00:00 2001 From: Yukang-Lian Date: Fri, 17 Apr 2026 19:53:51 +0800 Subject: [PATCH 2/3] [format](be) clang-format task_worker_pool.cpp after compaction callback changes Reflow the new submit_compaction_task calls and the cloud callback's log messages to satisfy clang-format. --- be/src/agent/task_worker_pool.cpp | 14 ++++++-------- 1 file changed, 6 insertions(+), 8 deletions(-) diff --git a/be/src/agent/task_worker_pool.cpp b/be/src/agent/task_worker_pool.cpp index bb974927ddec9c..97125ffe6a888f 100644 --- a/be/src/agent/task_worker_pool.cpp +++ b/be/src/agent/task_worker_pool.cpp @@ -1632,9 +1632,9 @@ void submit_table_compaction_callback(StorageEngine& engine, const TAgentTaskReq // compaction_action.cpp _handle_run_compaction). `force=false` keeps the // admission under permit limiter, matching the HTTP API default. tablet_ptr->set_last_full_compaction_schedule_time(UnixMillis()); - Status status = engine.submit_compaction_task( - tablet_ptr, CompactionType::FULL_COMPACTION, - /*force=*/false, /*eager=*/true, /*trigger_method=*/1); + Status status = engine.submit_compaction_task(tablet_ptr, CompactionType::FULL_COMPACTION, + /*force=*/false, /*eager=*/true, + /*trigger_method=*/1); if (!status.ok()) { LOG(WARNING) << "failed to submit full compaction task. tablet_id=" << tablet_ptr->tablet_id() << ", error=" << status; @@ -1681,8 +1681,7 @@ void cloud_submit_table_compaction_callback(CloudStorageEngine& engine, // delete bitmap synced eagerly, full does not (FullCompaction re-syncs itself). bool sync_delete_bitmap = compaction_type != CompactionType::FULL_COMPACTION; auto tablet_res = engine.tablet_mgr().get_tablet(compaction_req.tablet_id, - /*warmup_data=*/false, - sync_delete_bitmap); + /*warmup_data=*/false, sync_delete_bitmap); if (!tablet_res.has_value()) { LOG(WARNING) << "failed to get cloud tablet. tablet_id=" << compaction_req.tablet_id << ", error=" << tablet_res.error(); @@ -1711,9 +1710,8 @@ void cloud_submit_table_compaction_callback(CloudStorageEngine& engine, Status status = engine.submit_compaction_task(tablet, compaction_type, /*trigger_method=*/1); if (!status.ok()) { - LOG(WARNING) << "failed to submit cloud compaction task. tablet_id=" - << tablet->tablet_id() << ", type=" << compaction_req.type - << ", error=" << status; + LOG(WARNING) << "failed to submit cloud compaction task. tablet_id=" << tablet->tablet_id() + << ", type=" << compaction_req.type << ", error=" << status; } } From 95a878fed50101f2fb5c155efaf16914f36a3caa Mon Sep 17 00:00:00 2001 From: Yukang-Lian Date: Mon, 20 Apr 2026 11:52:44 +0800 Subject: [PATCH 3/3] [fix](cloud) address review feedback on CloudEnv.compactTable Two follow-ups on the AI review: 1. Do not hold the table read lock while resolving backends. CloudReplica.getBackendId() can call into Meta Service and wait for compute-group auto-start, which could block unrelated DDL on this table for the full wait. Split the method into two passes: collect tablet metadata (partition/index/tablet/schemaHash/replica refs) under the read lock, then release it and resolve backends outside. 2. Do not swallow UserException as a skip. The previous catch dropped real errors (missing USAGE privilege, manually shut down compute group, missing compute group, ...) to a LOG.warn, making failed ADMIN COMPACT TABLE look like a no-op. Now any UserException raised during backend resolution is surfaced to the user as DdlException. Compile-only verified; existing regression tests cover the happy path. --- .../apache/doris/cloud/catalog/CloudEnv.java | 75 +++++++++++++------ 1 file changed, 52 insertions(+), 23 deletions(-) diff --git a/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudEnv.java b/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudEnv.java index 69e050b9c298f3..f4393b36e7268c 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudEnv.java +++ b/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudEnv.java @@ -64,6 +64,7 @@ import java.io.DataInputStream; import java.io.File; import java.io.IOException; +import java.util.ArrayList; import java.util.List; import java.util.Map; import java.util.Optional; @@ -506,38 +507,31 @@ public void compactTable(String dbName, String tableName, String type, List pending = new ArrayList<>(); olapTable.readLock(); try { LOG.info("Cloud table compaction. db={}, table={}, partitions={}, type={}", dbName, tableName, partitionNames, type); + tableId = olapTable.getId(); for (String parName : partitionNames) { Partition partition = olapTable.getPartition(parName); if (partition == null) { throw new DdlException("partition[" + parName + "] not exist in table[" + tableName + "]"); } - for (MaterializedIndex idx : partition.getMaterializedIndices(IndexExtState.VISIBLE)) { + int schemaHash = olapTable.getSchemaHashByIndexId(idx.getId()); for (Tablet tablet : idx.getTablets()) { - // Cloud: each tablet has only one CloudReplica (primary BE). The backend - // resolution depends on the current compute group (ConnectContext); skip - // the tablet rather than aborting so other tablets can still dispatch. + // Cloud: each tablet has only one CloudReplica (primary BE). for (Replica replica : tablet.getReplicas()) { - long beId; - try { - beId = replica.getBackendId(); - } catch (UserException e) { - LOG.warn("skip tablet {} for cloud compaction, no available BE: {}", - tablet.getId(), e.getMessage()); - continue; - } - CompactionTask compactionTask = new CompactionTask( - beId, db.getId(), olapTable.getId(), partition.getId(), - idx.getId(), tablet.getId(), - olapTable.getSchemaHashByIndexId(idx.getId()), type); - batchTask.addTask(compactionTask); - dispatchedCount++; + pending.add(new PendingCloudCompactionTablet(partition.getId(), idx.getId(), + tablet.getId(), schemaHash, replica)); } } } @@ -546,10 +540,45 @@ public void compactTable(String dbName, String tableName, String type, List