From 521ab2742bfdaf2dcc4dbe4c36b950f2417db5c9 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Tue, 13 Oct 2026 19:56:05 +0800 Subject: [PATCH 1/2] [Feature] Expose DROP TABLE mod file progress --- .../schema/IoTDBDropTableModProgressIT.java | 105 ++++++++++++ .../relational/it/schema/IoTDBTableIT.java | 7 +- .../client/async/CnToDnAsyncRequestType.java | 1 + ...oDnInternalServiceAsyncRequestManager.java | 7 + .../rpc/DataNodeAsyncRequestRPCHandler.java | 9 + .../GetDropTableModProgressRPCHandler.java | 59 +++++++ .../confignode/manager/ProcedureManager.java | 11 ++ .../impl/schema/table/DropTableProcedure.java | 147 ++++++++++++++++- .../schema/table/DropTableProcedureTest.java | 124 ++++++++++++++ .../impl/DataNodeInternalRPCServiceImpl.java | 28 +++- ...formationSchemaContentSupplierFactory.java | 10 ++ .../storageengine/dataregion/DataRegion.java | 39 ++++- .../modification/DropTableModProgress.java | 65 ++++++++ .../write/RelationalDeleteDataNodeTest.java | 22 +++ .../dataregion/DataRegionTest.java | 154 +++++++++++++++++- .../schema/column/ColumnHeaderConstant.java | 2 + .../schema/table/InformationSchema.java | 6 + .../src/main/thrift/confignode.thrift | 2 + .../src/main/thrift/datanode.thrift | 19 +++ 19 files changed, 807 insertions(+), 10 deletions(-) create mode 100644 integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBDropTableModProgressIT.java create mode 100644 iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/rpc/GetDropTableModProgressRPCHandler.java create mode 100644 iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/modification/DropTableModProgress.java diff --git a/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBDropTableModProgressIT.java b/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBDropTableModProgressIT.java new file mode 100644 index 0000000000000..fe00da05a05c1 --- /dev/null +++ b/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBDropTableModProgressIT.java @@ -0,0 +1,105 @@ +/* + * 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. + */ + +package org.apache.iotdb.relational.it.schema; + +import org.apache.iotdb.db.it.utils.TestUtils; +import org.apache.iotdb.it.env.EnvFactory; +import org.apache.iotdb.it.framework.IoTDBTestRunner; +import org.apache.iotdb.itbase.category.TableClusterIT; +import org.apache.iotdb.itbase.category.TableLocalStandaloneIT; +import org.apache.iotdb.itbase.env.BaseEnv; + +import org.junit.AfterClass; +import org.junit.BeforeClass; +import org.junit.Test; +import org.junit.experimental.categories.Category; +import org.junit.runner.RunWith; + +import java.sql.Connection; +import java.sql.ResultSet; +import java.sql.Statement; +import java.util.Collections; +import java.util.concurrent.TimeUnit; + +import static org.awaitility.Awaitility.await; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +@RunWith(IoTDBTestRunner.class) +@Category({TableLocalStandaloneIT.class, TableClusterIT.class}) +public class IoTDBDropTableModProgressIT { + + @BeforeClass + public static void setUp() throws Exception { + EnvFactory.getEnv() + .getConfig() + .getCommonConfig() + .setEnforceStrongPassword(false) + .setDataRegionGroupExtensionPolicy("CUSTOM") + .setDefaultDataRegionGroupNumPerDatabase(1); + EnvFactory.getEnv().initClusterEnvironment(); + } + + @AfterClass + public static void tearDown() throws Exception { + EnvFactory.getEnv().cleanClusterEnvironment(); + } + + @Test + public void testDropTableModFileProgress() throws Exception { + try (final Connection connection = + EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT); + final Statement statement = connection.createStatement()) { + statement.execute("create database async_drop_mod_db"); + statement.execute("use async_drop_mod_db"); + statement.execute("create table drop_target (device string tag, reading int32)"); + statement.execute("create table retained (device string tag, reading int32)"); + // One region ensures both tables share a TsFile. Dropping just one table must write a mod + // file rather than delete that TsFile outright, and replicas must not multiply the count. + statement.execute("insert into drop_target(time, device, reading) values (1, 'd1', 10)"); + statement.execute("insert into retained(time, device, reading) values (1, 'd1', 20)"); + statement.execute("flush"); + statement.execute("drop table drop_target"); + await() + .atMost(60, TimeUnit.SECONDS) + .untilAsserted( + () -> { + try (final ResultSet resultSet = + statement.executeQuery( + "select state, written_mod_file_count, remaining_mod_file_count " + + "from information_schema.drop_table_procedures " + + "where database = 'async_drop_mod_db' and table_name = 'drop_target'")) { + assertTrue(resultSet.next()); + assertEquals("SUCCESS", resultSet.getString(1)); + assertEquals(1, resultSet.getLong(2)); + assertFalse(resultSet.wasNull()); + assertEquals(0, resultSet.getLong(3)); + assertFalse(resultSet.wasNull()); + assertFalse(resultSet.next()); + } + }); + TestUtils.assertResultSetEqual( + statement.executeQuery("select device, reading from retained"), + "device,reading,", + Collections.singleton("d1,20,")); + } + } +} diff --git a/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBTableIT.java b/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBTableIT.java index 4589ec5d49ae7..2b9b2d8ecce93 100644 --- a/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBTableIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBTableIT.java @@ -1250,7 +1250,8 @@ public void testAsyncDropTableProgress() throws Exception { Collections.emptySet()); try (final ResultSet resultSet = statement.executeQuery( - "select procedure_id, state, progress, error_message " + "select procedure_id, state, progress, error_message, " + + "written_mod_file_count, remaining_mod_file_count " + "from information_schema.drop_table_procedures " + "where database = 'async_drop_db' and table_name = 'drop_target'")) { assertTrue(resultSet.next()); @@ -1258,6 +1259,10 @@ public void testAsyncDropTableProgress() throws Exception { assertEquals("SUCCESS", resultSet.getString(2)); assertEquals("SUCCESS", resultSet.getString(3)); Assert.assertNull(resultSet.getString(4)); + assertEquals(0, resultSet.getLong(5)); + assertFalse(resultSet.wasNull()); + assertEquals(0, resultSet.getLong(6)); + assertFalse(resultSet.wasNull()); assertFalse(resultSet.next()); } } diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/CnToDnAsyncRequestType.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/CnToDnAsyncRequestType.java index 0749ba5f9b02e..3595644edf59b 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/CnToDnAsyncRequestType.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/CnToDnAsyncRequestType.java @@ -125,6 +125,7 @@ public enum CnToDnAsyncRequestType { UPDATE_TABLE, PRE_DELETE_TABLE, DELETE_DATA_FOR_DROP_TABLE, + GET_DROP_TABLE_MOD_PROGRESS, DELETE_DEVICES_FOR_DROP_TABLE, INVALIDATE_COLUMN_CACHE, DELETE_COLUMN_DATA, diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/CnToDnInternalServiceAsyncRequestManager.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/CnToDnInternalServiceAsyncRequestManager.java index e73573ea0e8a0..3010f8b524c04 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/CnToDnInternalServiceAsyncRequestManager.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/CnToDnInternalServiceAsyncRequestManager.java @@ -39,6 +39,7 @@ import org.apache.iotdb.confignode.client.async.handlers.rpc.DataNodeTSStatusRPCHandler; import org.apache.iotdb.confignode.client.async.handlers.rpc.FetchSchemaBlackListRPCHandler; import org.apache.iotdb.confignode.client.async.handlers.rpc.GetBuiltInExternalServiceRPCHandler; +import org.apache.iotdb.confignode.client.async.handlers.rpc.GetDropTableModProgressRPCHandler; import org.apache.iotdb.confignode.client.async.handlers.rpc.PipeHeartbeatRPCHandler; import org.apache.iotdb.confignode.client.async.handlers.rpc.PipePushMetaRPCHandler; import org.apache.iotdb.confignode.client.async.handlers.rpc.SchemaUpdateRPCHandler; @@ -78,6 +79,7 @@ import org.apache.iotdb.mpp.rpc.thrift.TDropPipePluginInstanceReq; import org.apache.iotdb.mpp.rpc.thrift.TDropTriggerInstanceReq; import org.apache.iotdb.mpp.rpc.thrift.TFetchSchemaBlackListReq; +import org.apache.iotdb.mpp.rpc.thrift.TGetDropTableModProgressReq; import org.apache.iotdb.mpp.rpc.thrift.TInactiveTriggerInstanceReq; import org.apache.iotdb.mpp.rpc.thrift.TInvalidateCacheReq; import org.apache.iotdb.mpp.rpc.thrift.TInvalidateColumnCacheReq; @@ -457,6 +459,11 @@ protected void initActionMapBuilder() { (req, client, handler) -> client.invalidateTableCache( (TInvalidateTableCacheReq) req, (DataNodeTSStatusRPCHandler) handler)); + actionMapBuilder.put( + CnToDnAsyncRequestType.GET_DROP_TABLE_MOD_PROGRESS, + (req, client, handler) -> + client.getDropTableModProgress( + (TGetDropTableModProgressReq) req, (GetDropTableModProgressRPCHandler) handler)); actionMapBuilder.put( CnToDnAsyncRequestType.DELETE_DATA_FOR_DROP_TABLE, (req, client, handler) -> diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/rpc/DataNodeAsyncRequestRPCHandler.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/rpc/DataNodeAsyncRequestRPCHandler.java index 7733d1f145ae6..cf4accb2f1157 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/rpc/DataNodeAsyncRequestRPCHandler.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/rpc/DataNodeAsyncRequestRPCHandler.java @@ -36,6 +36,7 @@ import org.apache.iotdb.mpp.rpc.thrift.TCountPathsUsingTemplateResp; import org.apache.iotdb.mpp.rpc.thrift.TDeviceViewResp; import org.apache.iotdb.mpp.rpc.thrift.TFetchSchemaBlackListResp; +import org.apache.iotdb.mpp.rpc.thrift.TGetDropTableModProgressResp; import org.apache.iotdb.mpp.rpc.thrift.TPullCommitProgressResp; import org.apache.iotdb.mpp.rpc.thrift.TPushConsumerGroupMetaResp; import org.apache.iotdb.mpp.rpc.thrift.TPushPipeMetaResp; @@ -82,6 +83,14 @@ public static DataNodeAsyncRequestRPCHandler buildHandler( Map responseMap = context.getResponseMap(); CountDownLatch countDownLatch = context.getCountDownLatch(); switch (requestType) { + case GET_DROP_TABLE_MOD_PROGRESS: + return new GetDropTableModProgressRPCHandler( + requestType, + requestId, + targetDataNode, + dataNodeLocationMap, + (Map) responseMap, + countDownLatch); case CONSTRUCT_SCHEMA_BLACK_LIST: case ROLLBACK_SCHEMA_BLACK_LIST: case DELETE_DATA_FOR_DELETE_SCHEMA: diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/rpc/GetDropTableModProgressRPCHandler.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/rpc/GetDropTableModProgressRPCHandler.java new file mode 100644 index 0000000000000..60eebcae2cfd7 --- /dev/null +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/rpc/GetDropTableModProgressRPCHandler.java @@ -0,0 +1,59 @@ +/* + * 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. + */ + +package org.apache.iotdb.confignode.client.async.handlers.rpc; + +import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation; +import org.apache.iotdb.confignode.client.async.CnToDnAsyncRequestType; +import org.apache.iotdb.mpp.rpc.thrift.TGetDropTableModProgressResp; +import org.apache.iotdb.rpc.RpcUtils; +import org.apache.iotdb.rpc.TSStatusCode; + +import java.util.Map; +import java.util.concurrent.CountDownLatch; + +public class GetDropTableModProgressRPCHandler + extends DataNodeAsyncRequestRPCHandler { + + public GetDropTableModProgressRPCHandler( + CnToDnAsyncRequestType requestType, + int requestId, + TDataNodeLocation targetNode, + Map nodeLocationMap, + Map responseMap, + CountDownLatch countDownLatch) { + super(requestType, requestId, targetNode, nodeLocationMap, responseMap, countDownLatch); + } + + @Override + public void onComplete(final TGetDropTableModProgressResp response) { + responseMap.put(requestId, response); + nodeLocationMap.remove(requestId); + countDownLatch.countDown(); + } + + @Override + public void onError(final Exception e) { + responseMap.put( + requestId, + new TGetDropTableModProgressResp( + RpcUtils.getStatus(TSStatusCode.CAN_NOT_CONNECT_DATANODE))); + countDownLatch.countDown(); + } +} diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ProcedureManager.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ProcedureManager.java index b9b84f20b658b..18f65a98481da 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ProcedureManager.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ProcedureManager.java @@ -171,6 +171,7 @@ import org.apache.iotdb.confignode.rpc.thrift.TUnsubscribeReq; import org.apache.iotdb.consensus.ConsensusFactory; import org.apache.iotdb.db.exception.BatchProcessException; +import org.apache.iotdb.mpp.rpc.thrift.TDropTableModProgress; import org.apache.iotdb.rpc.RpcUtils; import org.apache.iotdb.rpc.TSStatusCode; import org.apache.iotdb.rpc.subscription.config.TopicConfig; @@ -2549,6 +2550,9 @@ public List getDropTableProcedures() { && !(procedure instanceof DropViewProcedure)) .map(procedure -> (DropTableProcedure) procedure) .forEach(procedure -> dropProcedures.put(procedure.getProcId(), procedure)); + dropProcedures.values().stream() + .filter(procedure -> !procedure.isFinished()) + .forEach(DropTableProcedure::updateModFileProgress); return dropProcedures.values().stream() .map( procedure -> { @@ -2559,6 +2563,13 @@ public List getDropTableProcedures() { procedure.getProcId(), procedure.getState().name(), procedure.getDropProgress()); + final TDropTableModProgress modProgress = procedure.getModFileProgress(); + if (modProgress != null) { + info.setWrittenModFileCount(modProgress.getWrittenModFileCount()); + if (modProgress.isSetRemainingModFileCount()) { + info.setRemainingModFileCount(modProgress.getRemainingModFileCount()); + } + } if (procedure.hasException()) { info.setErrorMessage(procedure.getException().getMessage()); } diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/DropTableProcedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/DropTableProcedure.java index 64ea7a04be120..1aeb6954210fb 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/DropTableProcedure.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/DropTableProcedure.java @@ -43,6 +43,9 @@ import org.apache.iotdb.confignode.procedure.state.schema.DropTableState; import org.apache.iotdb.confignode.procedure.store.ProcedureType; import org.apache.iotdb.mpp.rpc.thrift.TDeleteDataOrDevicesForDropTableReq; +import org.apache.iotdb.mpp.rpc.thrift.TDropTableModProgress; +import org.apache.iotdb.mpp.rpc.thrift.TGetDropTableModProgressReq; +import org.apache.iotdb.mpp.rpc.thrift.TGetDropTableModProgressResp; import org.apache.iotdb.mpp.rpc.thrift.TInvalidateTableCacheReq; import org.apache.iotdb.rpc.TSStatusCode; @@ -51,13 +54,27 @@ import java.io.DataOutputStream; import java.io.IOException; +import java.nio.ByteBuffer; import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.locks.ReentrantLock; public class DropTableProcedure extends AbstractAlterOrDropTableProcedure { private static final Logger LOGGER = LoggerFactory.getLogger(DropTableProcedure.class); + // Track the replica actually executing each region, including changes caused by retries. + // Count each region once; followers may have different local TsFiles and apply asynchronously. + private final Map modProgressTargets = + new ConcurrentHashMap<>(); + private final ReentrantLock modProgressLock = new ReentrantLock(); + private volatile int modProgressRegionCount; + private volatile TDropTableModProgress modFileProgress; + private boolean modProgressFinished; + public DropTableProcedure(final boolean isGeneratedByPipe) { super(isGeneratedByPipe); } @@ -78,6 +95,90 @@ public String getDropProgress() { return state == null ? getState().name() : state.name(); } + public TDropTableModProgress getModFileProgress() { + final TDropTableModProgress progress = modFileProgress; + return progress == null ? null : progress.deepCopy(); + } + + /** Refreshes counters without allowing observability RPCs to block deletion indefinitely. */ + public void updateModFileProgress() { + if (!modProgressLock.tryLock()) { + return; + } + try { + refreshModFileProgress(); + } finally { + modProgressLock.unlock(); + } + } + + private void refreshModFileProgress() { + if (modProgressFinished + || isFinished() + || getCurrentState() != DropTableState.DELETE_DATA + || modProgressTargets.isEmpty() + || modProgressTargets.size() != modProgressRegionCount) { + return; + } + final DataNodeAsyncRequestContext + requestContext = + new DataNodeAsyncRequestContext<>(CnToDnAsyncRequestType.GET_DROP_TABLE_MOD_PROGRESS); + final Map> regionsByNode = new HashMap<>(); + final Map targets = new HashMap<>(modProgressTargets); + targets.forEach( + (region, location) -> { + requestContext.putNodeLocation(location.getDataNodeId(), location); + regionsByNode + .computeIfAbsent(location.getDataNodeId(), ignored -> new ArrayList<>()) + .add(region); + }); + for (final Map.Entry> entry : regionsByNode.entrySet()) { + requestContext.putRequest( + entry.getKey(), new TGetDropTableModProgressReq(getProcId(), entry.getValue())); + } + CnToDnInternalServiceAsyncRequestManager.getInstance() + .sendAsyncRequest(requestContext, 1, 2_000L, true); + + // A deletion retry may have selected a different replica while these RPCs were in flight. + if (targets.equals(modProgressTargets)) { + // A failed refresh is unknown, including the final refresh. Do not freeze stale counters + // with remaining files after deletion has finished. + modFileProgress = aggregateModFileProgress(targets, requestContext.getResponseMap()); + } + } + + static TDropTableModProgress aggregateModFileProgress( + final Map targets, + final Map responses) { + long written = 0; + long remaining = 0; + boolean hasUnknown = false; + for (final Map.Entry target : targets.entrySet()) { + final TGetDropTableModProgressResp response = + responses.get(target.getValue().getDataNodeId()); + if (response == null + || response.getStatus().getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode() + || !response.isSetRegionProgress() + || !response.getRegionProgress().containsKey(target.getKey().getId())) { + // Unknown is different from zero. In particular, an old DataNode does not report counts. + return null; + } + final TDropTableModProgress progress = + response.getRegionProgress().get(target.getKey().getId()); + written += progress.getWrittenModFileCount(); + if (progress.isSetRemainingModFileCount()) { + remaining += progress.getRemainingModFileCount(); + } else { + hasUnknown = true; + } + } + final TDropTableModProgress progress = new TDropTableModProgress(written); + if (!hasUnknown) { + progress.setRemainingModFileCount(remaining); + } + return progress; + } + @Override protected String getActionMessage() { return "drop table"; @@ -189,18 +290,34 @@ private void deleteData(final ConfigNodeProcedureEnv env) { env.getConfigManager().getRelatedDataRegionGroup4TableModel(database); if (!relatedDataRegionGroup.isEmpty()) { + modProgressRegionCount = relatedDataRegionGroup.size(); new TableRegionTaskExecutor<>( "delete data for drop table", env, relatedDataRegionGroup, CnToDnAsyncRequestType.DELETE_DATA_FOR_DROP_TABLE, - ((dataNodeLocation, consensusGroupIdList) -> - new TDeleteDataOrDevicesForDropTableReq( - new ArrayList<>(consensusGroupIdList), tableName))) + ((dataNodeLocation, consensusGroupIdList) -> { + consensusGroupIdList.forEach( + region -> modProgressTargets.put(region, dataNodeLocation)); + return new TDeleteDataOrDevicesForDropTableReq( + new ArrayList<>(consensusGroupIdList), tableName) + .setProcedureId(getProcId()); + })) .execute(); + // Serialize the final refresh with queries, using a separate lock so polling does not hold + // the procedure monitor or let an older query overwrite the final snapshot. + modProgressLock.lock(); + try { + refreshModFileProgress(); + modProgressFinished = true; + setNextState(DropTableState.DELETE_DEVICES); + } finally { + modProgressLock.unlock(); + } + } else { + modFileProgress = new TDropTableModProgress(0).setRemainingModFileCount(0); + setNextState(DropTableState.DELETE_DEVICES); } - - setNextState(DropTableState.DELETE_DEVICES); } private void deleteSchema(final ConfigNodeProcedureEnv env) { @@ -280,6 +397,26 @@ public void serialize(final DataOutputStream stream) throws IOException { ? ProcedureType.PIPE_ENRICHED_DROP_TABLE_PROCEDURE.getTypeCode() : ProcedureType.DROP_TABLE_PROCEDURE.getTypeCode()); super.serialize(stream); + final TDropTableModProgress progress = modFileProgress; + stream.writeBoolean(progress != null); + if (progress != null) { + stream.writeLong(progress.getWrittenModFileCount()); + stream.writeLong( + progress.isSetRemainingModFileCount() ? progress.getRemainingModFileCount() : -1); + } + } + + @Override + public void deserialize(final ByteBuffer buffer) { + super.deserialize(buffer); + if (!(this instanceof DropViewProcedure) && buffer.hasRemaining() && buffer.get() != 0) { + final TDropTableModProgress progress = new TDropTableModProgress(buffer.getLong()); + final long remaining = buffer.getLong(); + if (remaining >= 0) { + progress.setRemainingModFileCount(remaining); + } + modFileProgress = progress; + } } protected void innerSerialize(final DataOutputStream stream) throws IOException { diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/schema/table/DropTableProcedureTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/schema/table/DropTableProcedureTest.java index fb141aab270db..5d711be2deca2 100644 --- a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/schema/table/DropTableProcedureTest.java +++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/schema/table/DropTableProcedureTest.java @@ -19,7 +19,11 @@ package org.apache.iotdb.confignode.procedure.impl.schema.table; +import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId; +import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType; +import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation; import org.apache.iotdb.commons.exception.IllegalPathException; +import org.apache.iotdb.commons.utils.StatusUtils; import org.apache.iotdb.confignode.procedure.NoopProcedureStore; import org.apache.iotdb.confignode.procedure.Procedure; import org.apache.iotdb.confignode.procedure.ProcedureExecutor; @@ -28,6 +32,8 @@ import org.apache.iotdb.confignode.procedure.state.schema.DropTableState; import org.apache.iotdb.confignode.procedure.store.ProcedureType; import org.apache.iotdb.confignode.procedure.util.ProcedureTestUtil; +import org.apache.iotdb.mpp.rpc.thrift.TDropTableModProgress; +import org.apache.iotdb.mpp.rpc.thrift.TGetDropTableModProgressResp; import org.junit.Assert; import org.junit.Test; @@ -35,10 +41,12 @@ import java.io.ByteArrayOutputStream; import java.io.DataOutputStream; import java.io.IOException; +import java.lang.reflect.Field; import java.nio.ByteBuffer; import java.util.ArrayList; import java.util.Collections; import java.util.List; +import java.util.Map; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; @@ -225,5 +233,121 @@ public void serializeDeserializeTest() throws IllegalPathException, IOException deserializedProcedure.deserialize(byteBuffer); Assert.assertEquals(dropTableProcedure, deserializedProcedure); + Assert.assertNull(deserializedProcedure.getModFileProgress()); + Assert.assertFalse(byteBuffer.hasRemaining()); + } + + @Test + public void aggregateModProgressCountsEachRegionOnItsExecutingReplica() { + final TConsensusGroupId firstRegion = new TConsensusGroupId(TConsensusGroupType.DataRegion, 1); + final TConsensusGroupId secondRegion = new TConsensusGroupId(TConsensusGroupType.DataRegion, 2); + final Map targets = + Map.of( + firstRegion, new TDataNodeLocation().setDataNodeId(1), + secondRegion, new TDataNodeLocation().setDataNodeId(2)); + final Map responses = + Map.of( + 1, + new TGetDropTableModProgressResp(StatusUtils.OK) + .setRegionProgress( + Map.of(1, new TDropTableModProgress(2).setRemainingModFileCount(3))), + 2, + new TGetDropTableModProgressResp(StatusUtils.OK) + .setRegionProgress( + Map.of( + 1, new TDropTableModProgress(100).setRemainingModFileCount(100), + 2, new TDropTableModProgress(5).setRemainingModFileCount(7)))); + + Assert.assertEquals( + new TDropTableModProgress(7).setRemainingModFileCount(10), + DropTableProcedure.aggregateModFileProgress(targets, responses)); + } + + @Test + public void aggregateModProgressBeforeAllRegionsStart() { + final Map targets = + Map.of( + new TConsensusGroupId(TConsensusGroupType.DataRegion, 1), + new TDataNodeLocation().setDataNodeId(1), + new TConsensusGroupId(TConsensusGroupType.DataRegion, 2), + new TDataNodeLocation().setDataNodeId(1)); + final Map responses = + Map.of( + 1, + new TGetDropTableModProgressResp(StatusUtils.OK) + .setRegionProgress( + Map.of( + 1, + new TDropTableModProgress(2).setRemainingModFileCount(3), + 2, + new TDropTableModProgress(0)))); + + final TDropTableModProgress progress = + DropTableProcedure.aggregateModFileProgress(targets, responses); + Assert.assertEquals(2, progress.getWrittenModFileCount()); + Assert.assertFalse(progress.isSetRemainingModFileCount()); + } + + @Test + public void aggregateUnavailableModProgressIsUnknown() { + final Map targets = + Map.of( + new TConsensusGroupId(TConsensusGroupType.DataRegion, 1), + new TDataNodeLocation().setDataNodeId(1)); + Assert.assertNull(DropTableProcedure.aggregateModFileProgress(targets, Collections.emptyMap())); + Assert.assertNull( + DropTableProcedure.aggregateModFileProgress( + targets, Map.of(1, new TGetDropTableModProgressResp(StatusUtils.OK)))); + Assert.assertNull( + DropTableProcedure.aggregateModFileProgress( + targets, + Map.of( + 1, + new TGetDropTableModProgressResp(StatusUtils.OK) + .setRegionProgress(Collections.emptyMap())))); + } + + @Test + public void modProgressSurvivesSerialization() throws Exception { + for (final boolean isGeneratedByPipe : new boolean[] {false, true}) { + for (final TDropTableModProgress progress : + new TDropTableModProgress[] { + new TDropTableModProgress(4), + new TDropTableModProgress(4).setRemainingModFileCount(6), + new TDropTableModProgress(10).setRemainingModFileCount(0) + }) { + final DropTableProcedure procedure = + new DropTableProcedure("database", "table", "query", isGeneratedByPipe); + final Field progressField = DropTableProcedure.class.getDeclaredField("modFileProgress"); + progressField.setAccessible(true); + progressField.set(procedure, progress); + final ByteArrayOutputStream bytes = new ByteArrayOutputStream(); + procedure.serialize(new DataOutputStream(bytes)); + final ByteBuffer buffer = ByteBuffer.wrap(bytes.toByteArray()); + buffer.getShort(); + final DropTableProcedure recovered = new DropTableProcedure(isGeneratedByPipe); + recovered.deserialize(buffer); + Assert.assertEquals(progress, recovered.getModFileProgress()); + Assert.assertFalse(buffer.hasRemaining()); + // Callers cannot mutate the stored snapshot. + recovered.getModFileProgress().setWrittenModFileCount(0); + Assert.assertEquals(progress, recovered.getModFileProgress()); + } + } + } + + @Test + public void deserializeProcedureWithoutModProgress() throws Exception { + final DropTableProcedure procedure = + new DropTableProcedure("database", "table", "query", false); + final ByteArrayOutputStream bytes = new ByteArrayOutputStream(); + // The previous procedure format ended immediately after the superclass fields. + procedure.innerSerialize(new DataOutputStream(bytes)); + final ByteBuffer buffer = ByteBuffer.wrap(bytes.toByteArray()); + final DropTableProcedure recovered = new DropTableProcedure(false); + recovered.deserialize(buffer); + Assert.assertEquals(procedure, recovered); + Assert.assertNull(recovered.getModFileProgress()); + Assert.assertFalse(buffer.hasRemaining()); } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java index adce7c280fcd6..c8f2a0820a1f8 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java @@ -215,6 +215,7 @@ import org.apache.iotdb.db.storageengine.dataregion.compaction.settle.SettleRequestHandler; import org.apache.iotdb.db.storageengine.dataregion.flush.CompressionRatio; import org.apache.iotdb.db.storageengine.dataregion.modification.DeletionPredicate; +import org.apache.iotdb.db.storageengine.dataregion.modification.DropTableModProgress; import org.apache.iotdb.db.storageengine.dataregion.modification.TableDeletionEntry; import org.apache.iotdb.db.storageengine.dataregion.modification.TagPredicate; import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileManager; @@ -271,6 +272,7 @@ import org.apache.iotdb.mpp.rpc.thrift.TDeviceViewResp; import org.apache.iotdb.mpp.rpc.thrift.TDropFunctionInstanceReq; import org.apache.iotdb.mpp.rpc.thrift.TDropPipePluginInstanceReq; +import org.apache.iotdb.mpp.rpc.thrift.TDropTableModProgress; import org.apache.iotdb.mpp.rpc.thrift.TDropTriggerInstanceReq; import org.apache.iotdb.mpp.rpc.thrift.TExecuteCQ; import org.apache.iotdb.mpp.rpc.thrift.TFetchFragmentInstanceInfoReq; @@ -285,6 +287,8 @@ import org.apache.iotdb.mpp.rpc.thrift.TGenerateDataPartitionTableReq; import org.apache.iotdb.mpp.rpc.thrift.TGenerateDataPartitionTableResp; import org.apache.iotdb.mpp.rpc.thrift.TGetDataPartitionTableGeneratorProgressResp; +import org.apache.iotdb.mpp.rpc.thrift.TGetDropTableModProgressReq; +import org.apache.iotdb.mpp.rpc.thrift.TGetDropTableModProgressResp; import org.apache.iotdb.mpp.rpc.thrift.TGetEarliestTimeslotsResp; import org.apache.iotdb.mpp.rpc.thrift.TInactiveTriggerInstanceReq; import org.apache.iotdb.mpp.rpc.thrift.TInvalidateCacheReq; @@ -2009,7 +2013,9 @@ public TSStatus deleteDataForDropTable(final TDeleteDataOrDevicesForDropTableReq .execute( new DataRegionId(consensusGroupId.getId()), new RelationalDeleteDataNode( - new PlanNodeId(""), + req.isSetProcedureId() + ? DropTableModProgress.planNodeId(req.getProcedureId()) + : new PlanNodeId(""), new TableDeletionEntry( new DeletionPredicate(req.getTableName()), new TimeRange(Long.MIN_VALUE, Long.MAX_VALUE)), @@ -2018,6 +2024,26 @@ public TSStatus deleteDataForDropTable(final TDeleteDataOrDevicesForDropTableReq .getStatus()); } + @Override + public TGetDropTableModProgressResp getDropTableModProgress( + final TGetDropTableModProgressReq req) { + final Map progress = new HashMap<>(); + for (final TConsensusGroupId regionId : req.getRegionIdList()) { + final DataRegion region = + StorageEngine.getInstance().getDataRegion(new DataRegionId(regionId.getId())); + if (region != null) { + final TDropTableModProgress regionProgress = + region.getDropTableModProgress(req.getProcedureId()); + // Regions in the same deletion request execute sequentially. Report zero written and + // an unknown total until this region starts, so other regions' live counts are visible. + progress.put( + regionId.getId(), + regionProgress == null ? new TDropTableModProgress(0) : regionProgress); + } + } + return new TGetDropTableModProgressResp(StatusUtils.OK).setRegionProgress(progress); + } + @Override public TSStatus deleteDevicesForDropTable(final TDeleteDataOrDevicesForDropTableReq req) { return executeInternalSchemaTask( diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/InformationSchemaContentSupplierFactory.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/InformationSchemaContentSupplierFactory.java index 5b92f2593732d..bb8b0cf05e5db 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/InformationSchemaContentSupplierFactory.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/InformationSchemaContentSupplierFactory.java @@ -528,6 +528,16 @@ protected void constructLine() { } else { columnBuilders[5].appendNull(); } + if (currentProcedure.isSetWrittenModFileCount()) { + columnBuilders[6].writeLong(currentProcedure.getWrittenModFileCount()); + } else { + columnBuilders[6].appendNull(); + } + if (currentProcedure.isSetRemainingModFileCount()) { + columnBuilders[7].writeLong(currentProcedure.getRemainingModFileCount()); + } else { + columnBuilders[7].appendNull(); + } resultBuilder.declarePosition(); currentProcedure = null; } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java index ead75db6bac6e..deaefdd252673 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java @@ -123,6 +123,7 @@ import org.apache.iotdb.db.storageengine.dataregion.memtable.IMemTable; import org.apache.iotdb.db.storageengine.dataregion.memtable.TsFileProcessor; import org.apache.iotdb.db.storageengine.dataregion.memtable.TsFileProcessorInfo; +import org.apache.iotdb.db.storageengine.dataregion.modification.DropTableModProgress; import org.apache.iotdb.db.storageengine.dataregion.modification.ModEntry; import org.apache.iotdb.db.storageengine.dataregion.modification.ModificationFile; import org.apache.iotdb.db.storageengine.dataregion.modification.TableDeletionEntry; @@ -170,6 +171,7 @@ import org.apache.iotdb.db.utils.EncryptDBUtils; import org.apache.iotdb.db.utils.ModificationUtils; import org.apache.iotdb.metrics.utils.MetricLevel; +import org.apache.iotdb.mpp.rpc.thrift.TDropTableModProgress; import org.apache.iotdb.rpc.RpcUtils; import org.apache.iotdb.rpc.TSStatusCode; @@ -264,7 +266,6 @@ * will be called as a callback. */ public class DataRegion implements IDataRegionForQuery { - private static final IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig(); private static final Logger DEBUG_LOGGER = LoggerFactory.getLogger("QUERY_DEBUG"); @@ -290,6 +291,10 @@ public class DataRegion implements IDataRegionForQuery { /** Data region has been deleted or not. */ private volatile boolean deleted = false; + // Only the latest drop is retained per region. The ConfigNode captures its final counters + // before the dedicated DROP TABLE worker executes another procedure's DELETE_DATA state. + private volatile Pair dropTableModProgress; + /** closeStorageGroupCondition is used to wait for all currently closing TsFiles to be done. */ private final Object closeStorageGroupCondition = new Object(); @@ -3202,6 +3207,11 @@ public void deleteByDevice(final MeasurementPath pattern, final DeleteDataNode n } } + public TDropTableModProgress getDropTableModProgress(final long procedureId) { + final Pair progress = dropTableModProgress; + return progress != null && progress.left == procedureId ? progress.right.snapshot() : null; + } + public void deleteByTable(RelationalDeleteDataNode node) throws IOException { if (node.getDatabaseName() != null && !node.getDatabaseName().equals(databaseName)) { // not targeted on this database, return @@ -3215,12 +3225,24 @@ public void deleteByTable(RelationalDeleteDataNode node) throws IOException { } List modEntries = node.getModEntries(); + final long procedureId = DropTableModProgress.procedureId(node.getPlanNodeId()); + final DropTableModProgress progress = + procedureId >= 0 && modEntries.size() == 1 && modEntries.get(0).isDroppingTable() + ? new DropTableModProgress() + : null; + logger.info(StorageEngineMessages.DELETION_EXECUTING_TABLE_DELETION, node); writeLock("delete"); boolean hasReleasedLock = false; try { + if (progress != null) { + dropTableModProgress = new Pair<>(procedureId, progress); + } if (deleted) { + if (progress != null) { + progress.setTotalModFileCount(0); + } return; } String tableName = modEntries.get(0).getTableName(); @@ -3302,7 +3324,7 @@ public void deleteByTable(RelationalDeleteDataNode node) throws IOException { hasReleasedLock = true; for (int i = 0; i < modEntries.size(); i++) { - deleteDataInSealedFiles(sealedTsFileResourceLists.get(i), modEntries.get(i)); + deleteDataInSealedFiles(sealedTsFileResourceLists.get(i), modEntries.get(i), progress); } } catch (Exception e) { throw new IOException(e); @@ -3619,6 +3641,12 @@ private void deleteDataInUnsealedFiles( private void deleteDataInSealedFiles(Collection sealedTsFiles, ModEntry deletion) throws IOException { + deleteDataInSealedFiles(sealedTsFiles, deletion, null); + } + + private void deleteDataInSealedFiles( + Collection sealedTsFiles, ModEntry deletion, DropTableModProgress progress) + throws IOException { Set involvedModificationFiles = new HashSet<>(); List deletedByMods = new ArrayList<>(); List deletedByFiles = new ArrayList<>(); @@ -3753,6 +3781,10 @@ private void deleteDataInSealedFiles(Collection sealedTsFiles, M } } + if (progress != null) { + progress.setTotalModFileCount(involvedModificationFiles.size()); + } + if (involvedModificationFiles.isEmpty()) { logger.info(StorageEngineMessages.DELETION_DOES_NOT_INVOLVE_ANY_FILE, deletion); return; @@ -3765,6 +3797,9 @@ private void deleteDataInSealedFiles(Collection sealedTsFiles, M try { modFile.write(deletion); modFile.close(); + if (progress != null) { + progress.modFileWritten(); + } } catch (Exception e) { return e; } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/modification/DropTableModProgress.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/modification/DropTableModProgress.java new file mode 100644 index 0000000000000..88c9f019b5eb6 --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/modification/DropTableModProgress.java @@ -0,0 +1,65 @@ +/* + * 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. + */ + +package org.apache.iotdb.db.storageengine.dataregion.modification; + +import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId; +import org.apache.iotdb.mpp.rpc.thrift.TDropTableModProgress; + +/** Counts the distinct mod files processed by one region's current DROP TABLE attempt. */ +public class DropTableModProgress { + private static final String PLAN_NODE_ID_PREFIX = "drop-table-"; + + private long totalModFileCount = -1; + private long writtenModFileCount; + + public static PlanNodeId planNodeId(final long procedureId) { + // PlanNodeId already travels through consensus. Reuse it so observability does not change + // the deletion plan, WAL or DAL serialization formats. + return new PlanNodeId(PLAN_NODE_ID_PREFIX + procedureId); + } + + public static long procedureId(final PlanNodeId planNodeId) { + final String id = planNodeId.getId(); + if (!id.startsWith(PLAN_NODE_ID_PREFIX)) { + return -1; + } + try { + return Long.parseLong(id.substring(PLAN_NODE_ID_PREFIX.length())); + } catch (final NumberFormatException e) { + return -1; + } + } + + public synchronized void setTotalModFileCount(final long totalModFileCount) { + this.totalModFileCount = totalModFileCount; + } + + public synchronized void modFileWritten() { + ++writtenModFileCount; + } + + public synchronized TDropTableModProgress snapshot() { + final TDropTableModProgress progress = new TDropTableModProgress(writtenModFileCount); + if (totalModFileCount >= 0) { + progress.setRemainingModFileCount(totalModFileCount - writtenModFileCount); + } + return progress; + } +} diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/RelationalDeleteDataNodeTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/RelationalDeleteDataNodeTest.java index 910c38f44eb2d..bcbc490ea2a39 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/RelationalDeleteDataNodeTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/RelationalDeleteDataNodeTest.java @@ -24,6 +24,7 @@ import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId; import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeType; import org.apache.iotdb.db.storageengine.dataregion.modification.DeletionPredicate; +import org.apache.iotdb.db.storageengine.dataregion.modification.DropTableModProgress; import org.apache.iotdb.db.storageengine.dataregion.modification.TableDeletionEntry; import org.apache.iotdb.db.storageengine.dataregion.modification.TagPredicate.And; import org.apache.iotdb.db.storageengine.dataregion.modification.TagPredicate.FullExactMatch; @@ -47,6 +48,27 @@ public class RelationalDeleteDataNodeTest { + @Test + public void testDropTableProcedureIdSurvivesConsensusSerialization() throws IOException { + final RelationalDeleteDataNode node = + new RelationalDeleteDataNode( + DropTableModProgress.planNodeId(42), + new TableDeletionEntry( + new DeletionPredicate("table"), new TimeRange(Long.MIN_VALUE, Long.MAX_VALUE)), + null); + assertEquals( + 42, + DropTableModProgress.procedureId( + PlanNodeType.deserialize(node.serializeToByteBuffer()).getPlanNodeId())); + final ByteArrayOutputStream bytes = new ByteArrayOutputStream(); + node.serialize(new DataOutputStream(bytes)); + assertEquals( + 42, + DropTableModProgress.procedureId( + PlanNodeType.deserialize(ByteBuffer.wrap(bytes.toByteArray())).getPlanNodeId())); + assertEquals(-1, DropTableModProgress.procedureId(new PlanNodeId("query-plan"))); + } + @Test public void testSerialization() throws IOException { RelationalDeleteDataNode relationalDeleteDataNode = diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/DataRegionTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/DataRegionTest.java index 4e2ca24a7500e..b287a669b8081 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/DataRegionTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/DataRegionTest.java @@ -70,6 +70,8 @@ import org.apache.iotdb.db.storageengine.dataregion.memtable.ReadOnlyMemChunk; import org.apache.iotdb.db.storageengine.dataregion.memtable.TsFileProcessor; import org.apache.iotdb.db.storageengine.dataregion.modification.DeletionPredicate; +import org.apache.iotdb.db.storageengine.dataregion.modification.DropTableModProgress; +import org.apache.iotdb.db.storageengine.dataregion.modification.ModEntry; import org.apache.iotdb.db.storageengine.dataregion.modification.ModificationFile; import org.apache.iotdb.db.storageengine.dataregion.modification.TableDeletionEntry; import org.apache.iotdb.db.storageengine.dataregion.modification.TagPredicate.NOP; @@ -81,6 +83,7 @@ import org.apache.iotdb.db.storageengine.rescon.memory.SystemInfo; import org.apache.iotdb.db.utils.EnvironmentUtils; import org.apache.iotdb.db.utils.constant.TestConstant; +import org.apache.iotdb.mpp.rpc.thrift.TDropTableModProgress; import org.apache.iotdb.rpc.RpcUtils; import org.apache.iotdb.rpc.TSStatusCode; @@ -116,12 +119,15 @@ import java.util.concurrent.CompletableFuture; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import static org.apache.iotdb.db.queryengine.plan.statement.StatementTestUtils.genInsertRowNode; import static org.apache.iotdb.db.queryengine.plan.statement.StatementTestUtils.genInsertTabletNode; +import static org.awaitility.Awaitility.await; import static org.junit.Assert.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyBoolean; @@ -2234,7 +2240,11 @@ public void testTableDeletionFallsBackToModsForCompactingFile() throws Exception new TimeRange(Long.MIN_VALUE, Long.MAX_VALUE)); dataRegion.deleteByTable( new RelationalDeleteDataNode( - new PlanNodeId("table-delete"), deletionEntry, dataRegion.getDatabaseName())); + DropTableModProgress.planNodeId(42), deletionEntry, dataRegion.getDatabaseName())); + + Assert.assertEquals( + new TDropTableModProgress(1).setRemainingModFileCount(0), + dataRegion.getDropTableModProgress(42)); assertTrue(sourceTsFile.getTsFile().exists()); assertTrue(sourceTsFile.isCompacting()); @@ -2250,6 +2260,148 @@ public void testTableDeletionFallsBackToModsForCompactingFile() throws Exception } } + @Test + public void testDropTableModProgressWhileClosingSharedFile() throws Exception { + checkDropTableModProgressWhileClosingFile(true); + } + + @Test + public void testDropTableModProgressWhileClosingFile() throws Exception { + checkDropTableModProgressWhileClosingFile(false); + } + + private void checkDropTableModProgressWhileClosingFile(final boolean shareModFile) + throws Exception { + dataRegion.insertTablet(genInsertTabletNode(10, 0)); + dataRegion.syncCloseAllWorkingTsFileProcessors(); + dataRegion.insertTablet(genInsertTabletNode(10, 100)); + dataRegion.syncCloseAllWorkingTsFileProcessors(); + final List resources = dataRegion.getTsFileManager().getTsFileList(true); + Assert.assertEquals(2, resources.size()); + final CountDownLatch closeStarted = new CountDownLatch(1); + final CountDownLatch allowClose = new CountDownLatch(1); + final AtomicInteger writeCount = new AtomicInteger(); + final ModificationFile modFile = + new ModificationFile(resources.get(0).getModFileForWrite().getFile(), true) { + @Override + public int hashCode() { + // Separate HashSet buckets ensure the stream can split these two writes. + return 0; + } + + @Override + public void write(final ModEntry entry) throws IOException { + writeCount.incrementAndGet(); + super.write(entry); + } + + @Override + public void close() throws IOException { + closeStarted.countDown(); + try { + assertTrue(allowClose.await(10, TimeUnit.SECONDS)); + } catch (final InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IOException(e); + } + super.close(); + } + }; + for (final TsFileResource resource : resources) { + assertTrue(resource.transformStatus(TsFileResourceStatus.COMPACTION_CANDIDATE)); + assertTrue(resource.transformStatus(TsFileResourceStatus.COMPACTING)); + if (shareModFile || resource == resources.get(0)) { + resource.setExclusiveModFile(modFile); + } else { + resource.setExclusiveModFile( + new ModificationFile(resource.getModFileForWrite().getFile(), true) { + @Override + public int hashCode() { + return 15; + } + }); + } + } + + final ExecutorService executor = Executors.newSingleThreadExecutor(); + try { + final Future deletion = + executor.submit( + () -> { + dataRegion.deleteByTable(dropTableNode(43)); + return null; + }); + assertTrue(closeStarted.await(10, TimeUnit.SECONDS)); + Assert.assertFalse(deletion.isDone()); + // A successful write is still pending until close returns. With independent mod files, + // the other file finishes while this one is blocked. Shared mod files are counted once. + await() + .atMost(10, TimeUnit.SECONDS) + .untilAsserted( + () -> + Assert.assertEquals( + new TDropTableModProgress(shareModFile ? 0 : 1).setRemainingModFileCount(1), + dataRegion.getDropTableModProgress(43))); + allowClose.countDown(); + deletion.get(10, TimeUnit.SECONDS); + Assert.assertEquals( + new TDropTableModProgress(shareModFile ? 1 : 2).setRemainingModFileCount(0), + dataRegion.getDropTableModProgress(43)); + Assert.assertEquals(1, writeCount.get()); + } finally { + allowClose.countDown(); + executor.shutdown(); + assertTrue(executor.awaitTermination(10, TimeUnit.SECONDS)); + } + } + + @Test + public void testDropTableModProgressDoesNotCountFailedClose() throws Exception { + dataRegion.insertTablet(genInsertTabletNode(10, 0)); + dataRegion.syncCloseAllWorkingTsFileProcessors(); + final TsFileResource resource = dataRegion.getTsFileManager().getTsFileList(true).get(0); + assertTrue(resource.transformStatus(TsFileResourceStatus.COMPACTION_CANDIDATE)); + assertTrue(resource.transformStatus(TsFileResourceStatus.COMPACTING)); + final ModificationFile modFile = Mockito.spy(resource.getModFileForWrite()); + resource.setExclusiveModFile(modFile); + Mockito.doThrow(new IOException()).when(modFile).close(); + try { + Assert.assertThrows(IOException.class, () -> dataRegion.deleteByTable(dropTableNode(44))); + Assert.assertEquals( + new TDropTableModProgress(0).setRemainingModFileCount(1), + dataRegion.getDropTableModProgress(44)); + } finally { + Mockito.doCallRealMethod().when(modFile).close(); + modFile.close(); + } + } + + @Test + public void testDropTableModProgressWithoutModFiles() throws Exception { + dataRegion.deleteByTable(dropTableNode(45)); + Assert.assertEquals( + new TDropTableModProgress(0).setRemainingModFileCount(0), + dataRegion.getDropTableModProgress(45)); + dataRegion.insertTablet(genInsertTabletNode(10, 0)); + dataRegion.syncCloseAllWorkingTsFileProcessors(); + final TsFileResource resource = dataRegion.getTsFileManager().getTsFileList(true).get(0); + dataRegion.deleteByTable(dropTableNode(46)); + Assert.assertFalse(resource.getTsFile().exists()); + Assert.assertEquals( + new TDropTableModProgress(0).setRemainingModFileCount(0), + dataRegion.getDropTableModProgress(46)); + Assert.assertNull(dataRegion.getDropTableModProgress(45)); + } + + private RelationalDeleteDataNode dropTableNode(final long procedureId) { + return new RelationalDeleteDataNode( + DropTableModProgress.planNodeId(procedureId), + new TableDeletionEntry( + new DeletionPredicate(StatementTestUtils.tableName()), + new TimeRange(Long.MIN_VALUE, Long.MAX_VALUE)), + dataRegion.getDatabaseName()); + } + @Test public void testFlushSpecifiedResource() throws IllegalPathException, WriteProcessException, ExecutionException, InterruptedException { diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/column/ColumnHeaderConstant.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/column/ColumnHeaderConstant.java index 03ba898f77fad..116ab1eb386d7 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/column/ColumnHeaderConstant.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/column/ColumnHeaderConstant.java @@ -308,6 +308,8 @@ private ColumnHeaderConstant() { public static final String PROCEDURE_ID_TABLE_MODEL = "procedure_id"; public static final String PROGRESS_TABLE_MODEL = "progress"; public static final String ERROR_MESSAGE_TABLE_MODEL = "error_message"; + public static final String WRITTEN_MOD_FILE_COUNT_TABLE_MODEL = "written_mod_file_count"; + public static final String REMAINING_MOD_FILE_COUNT_TABLE_MODEL = "remaining_mod_file_count"; public static final String REMAINING_EVENT_COUNT_TABLE_MODEL = "remaining_event_count"; public static final String ESTIMATED_REMAINING_SECONDS_TABLE_MODEL = "estimated_remaining_seconds"; diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/table/InformationSchema.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/table/InformationSchema.java index 9f3bf0bb2df0c..f08a7f9939c4b 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/table/InformationSchema.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/table/InformationSchema.java @@ -165,6 +165,12 @@ public class InformationSchema { dropTableProceduresTable.addColumnSchema( new AttributeColumnSchema( ColumnHeaderConstant.ERROR_MESSAGE_TABLE_MODEL, TSDataType.STRING)); + dropTableProceduresTable.addColumnSchema( + new AttributeColumnSchema( + ColumnHeaderConstant.WRITTEN_MOD_FILE_COUNT_TABLE_MODEL, TSDataType.INT64)); + dropTableProceduresTable.addColumnSchema( + new AttributeColumnSchema( + ColumnHeaderConstant.REMAINING_MOD_FILE_COUNT_TABLE_MODEL, TSDataType.INT64)); dropTableProceduresTable.removeColumnSchema(TsTable.TIME_COLUMN_NAME); schemaTables.put(DROP_TABLE_PROCEDURES, dropTableProceduresTable); diff --git a/iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift b/iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift index e2f2595fa7dc7..0b733829d87b1 100644 --- a/iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift +++ b/iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift @@ -1326,6 +1326,8 @@ struct TDropTableProcedureInfo { 4: required string state 5: required string progress 6: optional string errorMessage + 7: optional i64 writtenModFileCount + 8: optional i64 remainingModFileCount } struct TDescTableResp { diff --git a/iotdb-protocol/thrift-datanode/src/main/thrift/datanode.thrift b/iotdb-protocol/thrift-datanode/src/main/thrift/datanode.thrift index 73bd95942f1b2..76c0860a24fa8 100644 --- a/iotdb-protocol/thrift-datanode/src/main/thrift/datanode.thrift +++ b/iotdb-protocol/thrift-datanode/src/main/thrift/datanode.thrift @@ -380,6 +380,23 @@ struct TDeleteColumnDataReq { struct TDeleteDataOrDevicesForDropTableReq { 1: required list regionIdList 2: required string tableName + 3: optional i64 procedureId +} + +struct TDropTableModProgress { + 1: required i64 writtenModFileCount + // Unset until the region has identified all mod files involved in this deletion. + 2: optional i64 remainingModFileCount +} + +struct TGetDropTableModProgressReq { + 1: required i64 procedureId + 2: required list regionIdList +} + +struct TGetDropTableModProgressResp { + 1: required common.TSStatus status + 2: optional map regionProgress } struct TTableDeviceDeletionWithPatternAndFilterReq { @@ -1356,6 +1373,8 @@ service IDataNodeRPCService { */ common.TSStatus deleteDataForDropTable(TDeleteDataOrDevicesForDropTableReq req) + TGetDropTableModProgressResp getDropTableModProgress(TGetDropTableModProgressReq req) + /** * Delete devices for drop table */ From f434aa0416990ce7e917cf6497a2b9b8b57d8df3 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Wed, 14 Oct 2026 18:40:31 +0800 Subject: [PATCH 2/2] [Fix] Wait for in-flight DROP TABLE mod progress refresh --- .../impl/schema/table/DropTableProcedure.java | 73 +++++++++++++------ .../schema/table/DropTableProcedureTest.java | 32 ++++++++ 2 files changed, 84 insertions(+), 21 deletions(-) diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/DropTableProcedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/DropTableProcedure.java index 1aeb6954210fb..cdb4d4b71c81d 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/DropTableProcedure.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/DropTableProcedure.java @@ -59,8 +59,8 @@ import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.locks.ReentrantLock; public class DropTableProcedure extends AbstractAlterOrDropTableProcedure { @@ -70,10 +70,11 @@ public class DropTableProcedure extends AbstractAlterOrDropTableProcedure modProgressTargets = new ConcurrentHashMap<>(); - private final ReentrantLock modProgressLock = new ReentrantLock(); + private final Object modProgressRefreshMonitor = new Object(); + private CompletableFuture modProgressRefreshFuture; private volatile int modProgressRegionCount; private volatile TDropTableModProgress modFileProgress; - private boolean modProgressFinished; + private volatile boolean modProgressFinished; public DropTableProcedure(final boolean isGeneratedByPipe) { super(isGeneratedByPipe); @@ -100,15 +101,46 @@ public TDropTableModProgress getModFileProgress() { return progress == null ? null : progress.deepCopy(); } - /** Refreshes counters without allowing observability RPCs to block deletion indefinitely. */ + /** Shares an in-flight counter refresh among concurrent observability requests. */ public void updateModFileProgress() { - if (!modProgressLock.tryLock()) { - return; - } - try { - refreshModFileProgress(); - } finally { - modProgressLock.unlock(); + runModProgressRefresh(this::refreshModFileProgress, false); + } + + private void runModProgressRefresh(final Runnable action, final boolean retryAfterWait) { + while (true) { + final CompletableFuture refreshFuture; + final boolean refreshOwner; + synchronized (modProgressRefreshMonitor) { + if (modProgressRefreshFuture == null) { + refreshFuture = new CompletableFuture<>(); + modProgressRefreshFuture = refreshFuture; + refreshOwner = true; + } else { + refreshFuture = modProgressRefreshFuture; + refreshOwner = false; + } + } + if (!refreshOwner) { + refreshFuture.join(); + if (retryAfterWait) { + continue; + } + return; + } + try { + action.run(); + return; + } catch (final RuntimeException | Error e) { + refreshFuture.completeExceptionally(e); + throw e; + } finally { + synchronized (modProgressRefreshMonitor) { + if (modProgressRefreshFuture == refreshFuture) { + modProgressRefreshFuture = null; + } + refreshFuture.complete(null); + } + } } } @@ -304,16 +336,15 @@ private void deleteData(final ConfigNodeProcedureEnv env) { .setProcedureId(getProcId()); })) .execute(); - // Serialize the final refresh with queries, using a separate lock so polling does not hold - // the procedure monitor or let an older query overwrite the final snapshot. - modProgressLock.lock(); - try { - refreshModFileProgress(); - modProgressFinished = true; - setNextState(DropTableState.DELETE_DEVICES); - } finally { - modProgressLock.unlock(); - } + // Wait for an in-flight query, then publish the final snapshot atomically with the state + // transition so a query cannot overwrite it after deletion finishes. + runModProgressRefresh( + () -> { + refreshModFileProgress(); + modProgressFinished = true; + setNextState(DropTableState.DELETE_DEVICES); + }, + true); } else { modFileProgress = new TDropTableModProgress(0).setRemainingModFileCount(0); setNextState(DropTableState.DELETE_DEVICES); diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/schema/table/DropTableProcedureTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/schema/table/DropTableProcedureTest.java index 5d711be2deca2..c74db40db14e6 100644 --- a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/schema/table/DropTableProcedureTest.java +++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/schema/table/DropTableProcedureTest.java @@ -47,6 +47,7 @@ import java.util.Collections; import java.util.List; import java.util.Map; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; @@ -307,6 +308,37 @@ public void aggregateUnavailableModProgressIsUnknown() { .setRegionProgress(Collections.emptyMap())))); } + @Test + public void concurrentModProgressQueriesWaitForInFlightRefresh() throws Exception { + final DropTableProcedure procedure = + new DropTableProcedure("database", "table", "query", false); + final Field futureField = DropTableProcedure.class.getDeclaredField("modProgressRefreshFuture"); + futureField.setAccessible(true); + final CountDownLatch waitingQueryStarted = new CountDownLatch(1); + final CompletableFuture refreshFuture = + new CompletableFuture() { + @Override + public Void join() { + waitingQueryStarted.countDown(); + return super.join(); + } + }; + futureField.set(procedure, refreshFuture); + final ExecutorService executor = Executors.newSingleThreadExecutor(); + try { + final Future waitingQuery = executor.submit(procedure::updateModFileProgress); + Assert.assertTrue(waitingQueryStarted.await(10, TimeUnit.SECONDS)); + Assert.assertFalse(waitingQuery.isDone()); + refreshFuture.complete(null); + waitingQuery.get(10, TimeUnit.SECONDS); + } finally { + refreshFuture.complete(null); + futureField.set(procedure, null); + executor.shutdownNow(); + Assert.assertTrue(executor.awaitTermination(10, TimeUnit.SECONDS)); + } + } + @Test public void modProgressSurvivesSerialization() throws Exception { for (final boolean isGeneratedByPipe : new boolean[] {false, true}) {