This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new db8bc17a27f [Feature] Expose DROP TABLE mod file progress (#18855)
db8bc17a27f is described below
commit db8bc17a27f0ab9927f52b37199b7e7e740fb997
Author: Caideyipi <[email protected]>
AuthorDate: Thu Oct 15 10:32:55 2026 +0800
[Feature] Expose DROP TABLE mod file progress (#18855)
* [Feature] Expose DROP TABLE mod file progress
* [Fix] Wait for in-flight DROP TABLE mod progress refresh
---
.../it/schema/IoTDBDropTableModProgressIT.java | 105 ++++++++++++
.../iotdb/relational/it/schema/IoTDBTableIT.java | 7 +-
.../client/async/CnToDnAsyncRequestType.java | 1 +
.../CnToDnInternalServiceAsyncRequestManager.java | 7 +
.../rpc/DataNodeAsyncRequestRPCHandler.java | 9 ++
.../rpc/GetDropTableModProgressRPCHandler.java | 59 +++++++
.../iotdb/confignode/manager/ProcedureManager.java | 11 ++
.../impl/schema/table/DropTableProcedure.java | 178 ++++++++++++++++++++-
.../impl/schema/table/DropTableProcedureTest.java | 156 ++++++++++++++++++
.../impl/DataNodeInternalRPCServiceImpl.java | 28 +++-
.../InformationSchemaContentSupplierFactory.java | 10 ++
.../db/storageengine/dataregion/DataRegion.java | 39 ++++-
.../modification/DropTableModProgress.java | 65 ++++++++
.../node/write/RelationalDeleteDataNodeTest.java | 22 +++
.../storageengine/dataregion/DataRegionTest.java | 154 +++++++++++++++++-
.../schema/column/ColumnHeaderConstant.java | 2 +
.../commons/schema/table/InformationSchema.java | 6 +
.../src/main/thrift/confignode.thrift | 2 +
.../src/main/thrift/datanode.thrift | 19 +++
19 files changed, 870 insertions(+), 10 deletions(-)
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 00000000000..fe00da05a05
--- /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 4589ec5d49a..2b9b2d8ecce 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 class IoTDBTableIT {
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 class IoTDBTableIT {
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 0749ba5f9b0..3595644edf5 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 e73573ea0e8..3010f8b524c 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.DataNodeAsyncReques
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.TDropFunctionInstanceReq;
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 @@ public class CnToDnInternalServiceAsyncRequestManager
(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 7733d1f145a..cf4accb2f11 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.TCheckTimeSeriesExistenceResp;
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 abstract class
DataNodeAsyncRequestRPCHandler<Response>
Map<Integer, ?> responseMap = context.getResponseMap();
CountDownLatch countDownLatch = context.getCountDownLatch();
switch (requestType) {
+ case GET_DROP_TABLE_MOD_PROGRESS:
+ return new GetDropTableModProgressRPCHandler(
+ requestType,
+ requestId,
+ targetDataNode,
+ dataNodeLocationMap,
+ (Map<Integer, TGetDropTableModProgressResp>) 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 00000000000..60eebcae2cf
--- /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<TGetDropTableModProgressResp> {
+
+ public GetDropTableModProgressRPCHandler(
+ CnToDnAsyncRequestType requestType,
+ int requestId,
+ TDataNodeLocation targetNode,
+ Map<Integer, TDataNodeLocation> nodeLocationMap,
+ Map<Integer, TGetDropTableModProgressResp> 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 b9b84f20b65..18f65a98481 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.TSubscribeReq;
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 class ProcedureManager {
&& !(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 class ProcedureManager {
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 64ea7a04be1..cdb4d4b71c8 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.impl.schema.table.view.DropViewProc
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,28 @@ import org.slf4j.LoggerFactory;
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.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
public class DropTableProcedure extends
AbstractAlterOrDropTableProcedure<DropTableState> {
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<TConsensusGroupId, TDataNodeLocation> modProgressTargets =
+ new ConcurrentHashMap<>();
+ private final Object modProgressRefreshMonitor = new Object();
+ private CompletableFuture<Void> modProgressRefreshFuture;
+ private volatile int modProgressRegionCount;
+ private volatile TDropTableModProgress modFileProgress;
+ private volatile boolean modProgressFinished;
+
public DropTableProcedure(final boolean isGeneratedByPipe) {
super(isGeneratedByPipe);
}
@@ -78,6 +96,121 @@ public class DropTableProcedure extends
AbstractAlterOrDropTableProcedure<DropTa
return state == null ? getState().name() : state.name();
}
+ public TDropTableModProgress getModFileProgress() {
+ final TDropTableModProgress progress = modFileProgress;
+ return progress == null ? null : progress.deepCopy();
+ }
+
+ /** Shares an in-flight counter refresh among concurrent observability
requests. */
+ public void updateModFileProgress() {
+ runModProgressRefresh(this::refreshModFileProgress, false);
+ }
+
+ private void runModProgressRefresh(final Runnable action, final boolean
retryAfterWait) {
+ while (true) {
+ final CompletableFuture<Void> 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);
+ }
+ }
+ }
+ }
+
+ private void refreshModFileProgress() {
+ if (modProgressFinished
+ || isFinished()
+ || getCurrentState() != DropTableState.DELETE_DATA
+ || modProgressTargets.isEmpty()
+ || modProgressTargets.size() != modProgressRegionCount) {
+ return;
+ }
+ final DataNodeAsyncRequestContext<TGetDropTableModProgressReq,
TGetDropTableModProgressResp>
+ requestContext =
+ new
DataNodeAsyncRequestContext<>(CnToDnAsyncRequestType.GET_DROP_TABLE_MOD_PROGRESS);
+ final Map<Integer, List<TConsensusGroupId>> regionsByNode = new
HashMap<>();
+ final Map<TConsensusGroupId, TDataNodeLocation> 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<Integer, List<TConsensusGroupId>> 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<TConsensusGroupId, TDataNodeLocation> targets,
+ final Map<Integer, TGetDropTableModProgressResp> responses) {
+ long written = 0;
+ long remaining = 0;
+ boolean hasUnknown = false;
+ for (final Map.Entry<TConsensusGroupId, TDataNodeLocation> 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 +322,33 @@ public class DropTableProcedure extends
AbstractAlterOrDropTableProcedure<DropTa
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();
+ // 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);
}
-
- setNextState(DropTableState.DELETE_DEVICES);
}
private void deleteSchema(final ConfigNodeProcedureEnv env) {
@@ -280,6 +428,26 @@ public class DropTableProcedure extends
AbstractAlterOrDropTableProcedure<DropTa
? 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 fb141aab270..c74db40db14 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.impl.schema.table.view.DropViewProc
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,13 @@ import org.junit.Test;
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.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@@ -225,5 +234,152 @@ public class DropTableProcedureTest {
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<TConsensusGroupId, TDataNodeLocation> targets =
+ Map.of(
+ firstRegion, new TDataNodeLocation().setDataNodeId(1),
+ secondRegion, new TDataNodeLocation().setDataNodeId(2));
+ final Map<Integer, TGetDropTableModProgressResp> 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<TConsensusGroupId, TDataNodeLocation> targets =
+ Map.of(
+ new TConsensusGroupId(TConsensusGroupType.DataRegion, 1),
+ new TDataNodeLocation().setDataNodeId(1),
+ new TConsensusGroupId(TConsensusGroupType.DataRegion, 2),
+ new TDataNodeLocation().setDataNodeId(1));
+ final Map<Integer, TGetDropTableModProgressResp> 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<TConsensusGroupId, TDataNodeLocation> 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 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<Void> refreshFuture =
+ new CompletableFuture<Void>() {
+ @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}) {
+ 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 adce7c280fc..c8f2a0820a1 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.schedule.Compacti
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.TDeviceViewReq;
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.TGenerateDataPartitionTableHeartbeatResp;
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 class DataNodeInternalRPCServiceImpl implements
IDataNodeRPCService.Iface
.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 class DataNodeInternalRPCServiceImpl implements
IDataNodeRPCService.Iface
.getStatus());
}
+ @Override
+ public TGetDropTableModProgressResp getDropTableModProgress(
+ final TGetDropTableModProgressReq req) {
+ final Map<Integer, TDropTableModProgress> 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 fb939fc5325..11a69b1ffca 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
@@ -532,6 +532,16 @@ public class InformationSchemaContentSupplierFactory {
} 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 ead75db6bac..deaefdd2526 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.flush.TsFileFlushPolicy;
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.CommonUtils;
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 @@ import static
org.apache.tsfile.common.constant.TsFileConstant.TSFILE_SUFFIX;
* 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<Long, DropTableModProgress> 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 class DataRegion implements IDataRegionForQuery {
}
}
+ public TDropTableModProgress getDropTableModProgress(final long procedureId)
{
+ final Pair<Long, DropTableModProgress> 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 class DataRegion implements IDataRegionForQuery {
}
List<TableDeletionEntry> 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 class DataRegion implements IDataRegionForQuery {
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 @@ public class DataRegion implements IDataRegionForQuery {
private void deleteDataInSealedFiles(Collection<TsFileResource>
sealedTsFiles, ModEntry deletion)
throws IOException {
+ deleteDataInSealedFiles(sealedTsFiles, deletion, null);
+ }
+
+ private void deleteDataInSealedFiles(
+ Collection<TsFileResource> sealedTsFiles, ModEntry deletion,
DropTableModProgress progress)
+ throws IOException {
Set<ModificationFile> involvedModificationFiles = new HashSet<>();
List<TsFileResource> deletedByMods = new ArrayList<>();
List<TsFileResource> deletedByFiles = new ArrayList<>();
@@ -3753,6 +3781,10 @@ public class DataRegion implements IDataRegionForQuery {
}
}
+ 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 @@ public class DataRegion implements IDataRegionForQuery {
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 00000000000..88c9f019b5e
--- /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 910c38f44eb..bcbc490ea2a 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.PlanNode;
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 @@ import static org.junit.Assert.assertEquals;
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 4e2ca24a750..b287a669b80 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.flush.TsFileFlushPolicy;
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.MemTableManager;
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.Set;
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 class DataRegionTest {
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 class DataRegionTest {
}
}
+ @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<TsFileResource> 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 56980f7e8df..d50be22154e 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 @@ public class 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 90e916027d4..3efaf7e1a9a 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
@@ -166,6 +166,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 e2f2595fa7d..0b733829d87 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 73bd95942f1..76c0860a24f 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<common.TConsensusGroupId> 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<common.TConsensusGroupId> regionIdList
+}
+
+struct TGetDropTableModProgressResp {
+ 1: required common.TSStatus status
+ 2: optional map<i32, TDropTableModProgress> regionProgress
}
struct TTableDeviceDeletionWithPatternAndFilterReq {
@@ -1356,6 +1373,8 @@ service IDataNodeRPCService {
*/
common.TSStatus deleteDataForDropTable(TDeleteDataOrDevicesForDropTableReq
req)
+ TGetDropTableModProgressResp
getDropTableModProgress(TGetDropTableModProgressReq req)
+
/**
* Delete devices for drop table
*/
