This is an automated email from the ASF dual-hosted git repository.
Caideyipi 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 27f89fa72fc [Feature] Make DROP TABLE asynchronous and expose deletion
progress (#18806)
27f89fa72fc is described below
commit 27f89fa72fc6f8ed4c478b8892bdc5fc669200b9
Author: Caideyipi <[email protected]>
AuthorDate: Tue Oct 13 14:16:56 2026 +0800
[Feature] Make DROP TABLE asynchronous and expose deletion progress (#18806)
* [Feature] Make DROP TABLE asynchronous and expose deletion progress
* [Test] Wait for asynchronous DROP TABLE completion in deletion IT
* [Fix] Address asynchronous DROP TABLE review comments
* [Fix] Disable RPC timeout for asynchronous DROP TABLE deletion
---
.../relational/it/db/it/IoTDBDeletionTableIT.java | 20 +++
.../relational/it/schema/IoTDBDatabaseIT.java | 4 +-
.../iotdb/relational/it/schema/IoTDBTableIT.java | 66 ++++++++
.../CnToDnInternalServiceAsyncRequestManager.java | 8 +-
.../iotdb/confignode/manager/ConfigManager.java | 10 ++
.../apache/iotdb/confignode/manager/IManager.java | 3 +
.../iotdb/confignode/manager/ProcedureManager.java | 99 ++++++++++--
.../confignode/procedure/ProcedureExecutor.java | 106 +++++++++---
.../procedure/TimeoutExecutorThread.java | 2 +-
.../impl/schema/table/DropTableProcedure.java | 8 +
.../thrift/ConfigNodeRPCServiceProcessor.java | 6 +
...ToDnInternalServiceAsyncRequestManagerTest.java | 114 +++++++++++++
.../impl/schema/table/DropTableProcedureTest.java | 177 +++++++++++++++++++++
.../iotdb/db/protocol/client/ConfigNodeClient.java | 7 +
.../InformationSchemaContentSupplierFactory.java | 68 ++++++++
.../DataNodeLocationSupplierFactory.java | 1 +
.../commons/client/AbstractConfigNodeClient.java | 6 +
.../schema/column/ColumnHeaderConstant.java | 3 +
.../commons/schema/table/InformationSchema.java | 20 +++
.../src/main/thrift/confignode.thrift | 15 ++
20 files changed, 711 insertions(+), 32 deletions(-)
diff --git
a/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBDeletionTableIT.java
b/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBDeletionTableIT.java
index f8732645926..c6063a2c59d 100644
---
a/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBDeletionTableIT.java
+++
b/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBDeletionTableIT.java
@@ -29,6 +29,7 @@ import org.apache.iotdb.itbase.category.ManualIT;
import org.apache.iotdb.itbase.category.TableClusterIT;
import org.apache.iotdb.itbase.category.TableLocalStandaloneIT;
import org.apache.iotdb.itbase.env.BaseEnv;
+import org.apache.iotdb.itbase.exception.InconsistentDataException;
import org.apache.iotdb.itbase.exception.ParallelRequestTimeoutException;
import org.apache.iotdb.rpc.IoTDBConnectionException;
import org.apache.iotdb.rpc.StatementExecutionException;
@@ -1361,6 +1362,25 @@ public class IoTDBDeletionTableIT {
statement.execute("DROP TABLE vehicle" + testNum);
+ // DROP TABLE acknowledges submission before the table has been fully
removed.
+ Awaitility.await()
+ .atMost(60, TimeUnit.SECONDS)
+ .pollInSameThread()
+ // Progress snapshots can differ between cluster nodes while
deletion is running.
+ .ignoreException(InconsistentDataException.class)
+ .untilAsserted(
+ () -> {
+ try (ResultSet resultSet =
+ statement.executeQuery(
+ "SELECT state, error_message FROM
information_schema.drop_table_procedures "
+ + "WHERE database = 'test' AND table_name =
'vehicle"
+ + testNum
+ + "' ORDER BY procedure_id DESC LIMIT 1")) {
+ assertTrue(resultSet.next());
+ assertEquals(resultSet.getString(2), "SUCCESS",
resultSet.getString(1));
+ }
+ });
+
try (ResultSet ignored = statement.executeQuery("SELECT * FROM vehicle"
+ testNum)) {
fail("Exception expected");
} catch (SQLException e) {
diff --git
a/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBDatabaseIT.java
b/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBDatabaseIT.java
index be42be99298..3fc22722078 100644
---
a/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBDatabaseIT.java
+++
b/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBDatabaseIT.java
@@ -532,6 +532,7 @@ public class IoTDBDatabaseIT {
"current_queries,INF,",
"data_nodes,INF,",
"databases,INF,",
+ "drop_table_procedures,INF,",
"functions,INF,",
"keywords,INF,",
"nodes,INF,",
@@ -829,6 +830,7 @@ public class IoTDBDatabaseIT {
Arrays.asList(
"information_schema,databases,INF,USING,null,SYSTEM
VIEW,false,",
"information_schema,tables,INF,USING,null,SYSTEM
VIEW,false,",
+
"information_schema,drop_table_procedures,INF,USING,null,SYSTEM VIEW,false,",
"information_schema,columns,INF,USING,null,SYSTEM
VIEW,false,",
"information_schema,queries,INF,USING,null,SYSTEM
VIEW,false,",
"information_schema,regions,INF,USING,null,SYSTEM
VIEW,false,",
@@ -854,7 +856,7 @@ public class IoTDBDatabaseIT {
TestUtils.assertResultSetEqual(
statement.executeQuery("count devices from tables where status =
'USING'"),
"count(devices),",
- Collections.singleton("24,"));
+ Collections.singleton("25,"));
TestUtils.assertResultSetEqual(
statement.executeQuery(
"select * from columns where table_name = 'queries' or database
= 'test'"),
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 4c13f660972..4589ec5d49a 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
@@ -546,6 +546,7 @@ public class IoTDBTableIT {
Collections.singleton("null,"));
statement.execute("drop table table2");
+ awaitDropProcedure(statement, "test2", "table2");
try {
statement.executeQuery("describe table2");
fail();
@@ -1229,4 +1230,69 @@ public class IoTDBTableIT {
}
}
}
+
+ @Test
+ public void testAsyncDropTableProgress() throws Exception {
+ try (final Connection connection =
+ EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ final Statement statement = connection.createStatement()) {
+ statement.execute("create database async_drop_db");
+ statement.execute("use async_drop_db");
+ statement.execute("create table drop_target (device string tag, reading
int32)");
+ statement.execute("drop table drop_target");
+
+ // The drop may finish before the first query. Its completed result
remains visible until
+ // the procedure cleaner evicts it, so verify that result after waiting
for completion.
+ awaitDropProcedure(statement, "async_drop_db", "drop_target");
+ TestUtils.assertResultSetEqual(
+ statement.executeQuery("show tables from async_drop_db"),
+ "TableName,TTL(ms),",
+ Collections.emptySet());
+ try (final ResultSet resultSet =
+ statement.executeQuery(
+ "select procedure_id, state, progress, error_message "
+ + "from information_schema.drop_table_procedures "
+ + "where database = 'async_drop_db' and table_name =
'drop_target'")) {
+ assertTrue(resultSet.next());
+ assertTrue(resultSet.getLong(1) >= 0);
+ assertEquals("SUCCESS", resultSet.getString(2));
+ assertEquals("SUCCESS", resultSet.getString(3));
+ Assert.assertNull(resultSet.getString(4));
+ assertFalse(resultSet.next());
+ }
+ }
+ }
+
+ private static void awaitDropProcedure(
+ final Statement statement, final String database, final String tableName)
+ throws SQLException {
+ final long deadline = System.currentTimeMillis() + 60_000;
+ while (System.currentTimeMillis() < deadline) {
+ try (final ResultSet resultSet =
+ statement.executeQuery(
+ "select state, error_message from
information_schema.drop_table_procedures "
+ + "where database = '"
+ + database
+ + "' and table_name = '"
+ + tableName
+ + "'")) {
+ if (resultSet.next()) {
+ final String state = resultSet.getString(1);
+ if ("SUCCESS".equals(state)) {
+ return;
+ }
+ if ("ROLLEDBACK".equals(state)) {
+ throw new SQLException(resultSet.getString(2));
+ }
+ }
+ }
+ try {
+ Thread.sleep(20);
+ } catch (final InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new SQLException(e);
+ }
+ }
+ throw new SQLException("Timed out waiting for drop table procedure");
+ }
}
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 7ca698afa2c..e73573ea0e8 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
@@ -558,11 +558,15 @@ public class CnToDnInternalServiceAsyncRequestManager
@Override
protected void adjustClientTimeoutIfNecessary(
- CnToDnAsyncRequestType CnToDnAsyncRequestType,
+ CnToDnAsyncRequestType requestType,
AsyncDataNodeInternalServiceClient client,
Long timeoutInMs) {
- if
(CnToDnAsyncRequestType.SUBMIT_TEST_CONNECTION_TASK.equals(CnToDnAsyncRequestType))
{
+ if
(CnToDnAsyncRequestType.SUBMIT_TEST_CONNECTION_TASK.equals(requestType)) {
client.setTimeoutTemporarily(TestConnectionUtils.calculateCnLeaderToAllDnMaxTime());
+ } else if
(CnToDnAsyncRequestType.DELETE_DATA_FOR_DROP_TABLE.equals(requestType)
+ ||
CnToDnAsyncRequestType.DELETE_DEVICES_FOR_DROP_TABLE.equals(requestType)) {
+ // Asynchronous DROP TABLE deletion can take arbitrarily long to finish.
+ client.setTimeoutTemporarily(-1);
} else if (timeoutInMs != null) {
client.setTimeoutTemporarily(timeoutInMs);
}
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java
index 98c0a1bb9fc..f82d8187fab 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java
@@ -243,6 +243,7 @@ import
org.apache.iotdb.confignode.rpc.thrift.TShowConfigNodesResp;
import
org.apache.iotdb.confignode.rpc.thrift.TShowDataNodes4InformationSchemaResp;
import org.apache.iotdb.confignode.rpc.thrift.TShowDataNodesResp;
import org.apache.iotdb.confignode.rpc.thrift.TShowDatabaseResp;
+import org.apache.iotdb.confignode.rpc.thrift.TShowDropTableProceduresResp;
import org.apache.iotdb.confignode.rpc.thrift.TShowPipePluginReq;
import org.apache.iotdb.confignode.rpc.thrift.TShowPipeReq;
import org.apache.iotdb.confignode.rpc.thrift.TShowPipeResp;
@@ -3401,6 +3402,15 @@ public class ConfigManager implements IManager {
: new TShowTable4InformationSchemaResp(status);
}
+ @Override
+ public TShowDropTableProceduresResp showDropTableProcedures() {
+ final TSStatus status = confirmLeader();
+ return status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()
+ ? new TShowDropTableProceduresResp(status)
+
.setDropTableProcedureInfoList(procedureManager.getDropTableProcedures())
+ : new TShowDropTableProceduresResp(status);
+ }
+
@Override
public TDescTableResp describeTable(
final String database, final String tableName, final boolean isDetails) {
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/IManager.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/IManager.java
index cc203caeeba..37cc7c72247 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/IManager.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/IManager.java
@@ -151,6 +151,7 @@ import
org.apache.iotdb.confignode.rpc.thrift.TShowConfigNodesResp;
import
org.apache.iotdb.confignode.rpc.thrift.TShowDataNodes4InformationSchemaResp;
import org.apache.iotdb.confignode.rpc.thrift.TShowDataNodesResp;
import org.apache.iotdb.confignode.rpc.thrift.TShowDatabaseResp;
+import org.apache.iotdb.confignode.rpc.thrift.TShowDropTableProceduresResp;
import org.apache.iotdb.confignode.rpc.thrift.TShowPipePluginReq;
import org.apache.iotdb.confignode.rpc.thrift.TShowPipeReq;
import org.apache.iotdb.confignode.rpc.thrift.TShowPipeResp;
@@ -924,6 +925,8 @@ public interface IManager {
TShowTable4InformationSchemaResp showTables4InformationSchema();
+ TShowDropTableProceduresResp showDropTableProcedures();
+
TDescTableResp describeTable(
final String database, final String tableName, final boolean isDetails);
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 9531f31f1b8..b9b84f20b65 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
@@ -32,12 +32,14 @@ import org.apache.iotdb.commons.conf.CommonDescriptor;
import org.apache.iotdb.commons.conf.IoTDBConstant;
import org.apache.iotdb.commons.exception.IoTDBException;
import org.apache.iotdb.commons.exception.MetadataException;
+import org.apache.iotdb.commons.exception.table.TableNotExistsException;
import org.apache.iotdb.commons.path.MeasurementPath;
import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.commons.path.PathDeserializeUtil;
import org.apache.iotdb.commons.path.PathPatternTree;
import org.apache.iotdb.commons.pipe.agent.plugin.meta.PipePluginMeta;
import org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant;
+import org.apache.iotdb.commons.schema.table.TableNodeStatus;
import org.apache.iotdb.commons.schema.table.TreeViewSchema;
import org.apache.iotdb.commons.schema.table.TsTable;
import org.apache.iotdb.commons.schema.table.column.TsTableColumnSchemaUtil;
@@ -60,6 +62,7 @@ import
org.apache.iotdb.confignode.consensus.request.write.region.CreateRegionGr
import org.apache.iotdb.confignode.i18n.ManagerMessages;
import org.apache.iotdb.confignode.i18n.ProcedureMessages;
import org.apache.iotdb.confignode.manager.partition.PartitionManager;
+import org.apache.iotdb.confignode.manager.schema.ClusterSchemaManager;
import
org.apache.iotdb.confignode.manager.subscription.SubscriptionCoordinator;
import org.apache.iotdb.confignode.persistence.ProcedureInfo;
import org.apache.iotdb.confignode.procedure.PartitionTableAutoCleaner;
@@ -159,6 +162,7 @@ import
org.apache.iotdb.confignode.rpc.thrift.TDeleteLogicalViewReq;
import org.apache.iotdb.confignode.rpc.thrift.TDeleteTableDeviceReq;
import org.apache.iotdb.confignode.rpc.thrift.TDeleteTableDeviceResp;
import org.apache.iotdb.confignode.rpc.thrift.TDropPipePluginReq;
+import org.apache.iotdb.confignode.rpc.thrift.TDropTableProcedureInfo;
import org.apache.iotdb.confignode.rpc.thrift.TExtendRegionReq;
import org.apache.iotdb.confignode.rpc.thrift.TMigrateRegionReq;
import org.apache.iotdb.confignode.rpc.thrift.TReconstructRegionReq;
@@ -2378,15 +2382,63 @@ public class ProcedureManager {
public TSStatus dropTable(final TAlterOrDropTableReq req) {
final boolean isView = req.isSetIsView() && req.isIsView();
- return executeWithoutDuplicate(
- req.database,
- null,
- req.tableName,
- req.queryId,
- isView ? ProcedureType.DROP_VIEW_PROCEDURE :
ProcedureType.DROP_TABLE_PROCEDURE,
- isView
- ? new DropViewProcedure(req.database, req.tableName, req.queryId,
false)
- : new DropTableProcedure(req.database, req.tableName, req.queryId,
false));
+ if (isView) {
+ return executeWithoutDuplicate(
+ req.database,
+ null,
+ req.tableName,
+ req.queryId,
+ ProcedureType.DROP_VIEW_PROCEDURE,
+ new DropViewProcedure(req.database, req.tableName, req.queryId,
false));
+ }
+ synchronized (this) {
+ if (!executor.isRunning()) {
+ return RpcUtils.getStatus(TSStatusCode.CONFIG_NODE_LEADER_WARMING_UP);
+ }
+ final Pair<Long, Boolean> duplicate =
+ checkDuplicateTableTask(
+ req.database,
+ null,
+ req.tableName,
+ null,
+ req.queryId,
+ ProcedureType.DROP_TABLE_PROCEDURE);
+ if (duplicate.getLeft() != -1) {
+ return StatusUtils.OK;
+ }
+ if (Boolean.TRUE.equals(duplicate.getRight())) {
+ return RpcUtils.getStatus(TSStatusCode.OVERLAP_WITH_EXISTING_TASK);
+ }
+ try {
+ final Optional<Pair<TsTable, TableNodeStatus>> table =
+ configManager
+ .getClusterSchemaManager()
+ .getTableAndStatusIfExists(req.database, req.tableName);
+ if (table.isEmpty()) {
+ final TableNotExistsException e =
+ new TableNotExistsException(req.database, req.tableName);
+ return RpcUtils.getStatus(e.getErrorCode(), e.getMessage());
+ }
+ if (table.get().getRight() == TableNodeStatus.PRE_DELETE) {
+ return RpcUtils.getStatus(TSStatusCode.OVERLAP_WITH_EXISTING_TASK);
+ }
+ final Optional<Pair<TSStatus, TsTable>> viewCheck =
+ ClusterSchemaManager.checkTable4View(req.database,
table.get().getLeft(), false);
+ if (viewCheck.isPresent()) {
+ return viewCheck.get().getLeft();
+ }
+ } catch (final MetadataException e) {
+ return RpcUtils.getStatus(e.getErrorCode(), e.getMessage());
+ }
+ final Procedure<ConfigNodeProcedureEnv> procedure =
+ new DropTableProcedure(req.database, req.tableName, req.queryId,
false);
+ try {
+ executor.submitProcedureAndPersist(procedure);
+ return StatusUtils.OK;
+ } catch (final Exception e) {
+ return RpcUtils.getStatus(TSStatusCode.INTERNAL_SERVER_ERROR,
e.getMessage());
+ }
+ }
}
public TSStatus renameTable(final TAlterOrDropTableReq req) {
@@ -2486,6 +2538,35 @@ public class ProcedureManager {
return result;
}
+ /** Returns running drops and completed results retained by the executor's
shared TTL cleaner. */
+ public List<TDropTableProcedureInfo> getDropTableProcedures() {
+ final Map<Long, DropTableProcedure> dropProcedures = new HashMap<>();
+ Stream.concat(
+ executor.getProcedures().values().stream(),
executor.getCompletedProcedures().stream())
+ .filter(
+ procedure ->
+ procedure instanceof DropTableProcedure
+ && !(procedure instanceof DropViewProcedure))
+ .map(procedure -> (DropTableProcedure) procedure)
+ .forEach(procedure -> dropProcedures.put(procedure.getProcId(),
procedure));
+ return dropProcedures.values().stream()
+ .map(
+ procedure -> {
+ final TDropTableProcedureInfo info =
+ new TDropTableProcedureInfo(
+ procedure.getDatabase(),
+ procedure.getTableName(),
+ procedure.getProcId(),
+ procedure.getState().name(),
+ procedure.getDropProgress());
+ if (procedure.hasException()) {
+ info.setErrorMessage(procedure.getException().getMessage());
+ }
+ return info;
+ })
+ .collect(Collectors.toList());
+ }
+
public TSStatus executeWithoutDuplicate(
final String database,
final TsTable table,
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/ProcedureExecutor.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/ProcedureExecutor.java
index dbbf33325a9..82b217b93ea 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/ProcedureExecutor.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/ProcedureExecutor.java
@@ -24,6 +24,8 @@ import org.apache.iotdb.commons.utils.RetryUtils;
import org.apache.iotdb.commons.utils.TestOnly;
import org.apache.iotdb.confignode.i18n.ProcedureMessages;
import org.apache.iotdb.confignode.procedure.exception.ProcedureException;
+import
org.apache.iotdb.confignode.procedure.impl.schema.table.DropTableProcedure;
+import
org.apache.iotdb.confignode.procedure.impl.schema.table.view.DropViewProcedure;
import org.apache.iotdb.confignode.procedure.scheduler.ProcedureScheduler;
import
org.apache.iotdb.confignode.procedure.scheduler.SimpleProcedureScheduler;
import org.apache.iotdb.confignode.procedure.state.ProcedureLockState;
@@ -71,6 +73,7 @@ public class ProcedureExecutor<Env> {
// Metrics may be scraped before init() and concurrently with initialization
during a ConfigNode
// leader transition.
private volatile CopyOnWriteArrayList<WorkerThread> workerThreads;
+ private volatile CopyOnWriteArrayList<WorkerThread> dropTableWorkerThreads;
private TimeoutExecutorThread<Env> timeoutExecutor;
@@ -89,6 +92,9 @@ public class ProcedureExecutor<Env> {
private CompletedProcedureRecycler<Env> completedProcedureRecycler;
private final ProcedureScheduler scheduler;
+ // Drop table can spend a long time deleting data on DataNodes. Keep its
worker separate so
+ // unrelated procedures can still make progress.
+ private final ProcedureScheduler dropTableScheduler = new
SimpleProcedureScheduler();
private final AtomicLong workId = new AtomicLong(0);
private final AtomicInteger activeExecutorCount = new AtomicInteger(0);
@@ -122,7 +128,7 @@ public class ProcedureExecutor<Env> {
return PROCEDURE_EXECUTION_CONTEXT.get();
}
- public void init(int numThreads) {
+ public synchronized void init(int numThreads) {
this.corePoolSize = numThreads;
this.maxPoolSize = 10 * numThreads;
this.timeoutExecutor =
@@ -134,12 +140,15 @@ public class ProcedureExecutor<Env> {
workId.set(0);
workerThreads = new CopyOnWriteArrayList<>();
for (int i = 0; i < corePoolSize; i++) {
- workerThreads.add(new WorkerThread(threadGroup));
+ workerThreads.add(new WorkerThread(threadGroup, scheduler));
}
+ dropTableWorkerThreads = new CopyOnWriteArrayList<>();
// Add worker monitor
workerMonitorExecutor.add(new WorkerMonitor());
+ dropTableScheduler.clear();
scheduler.start();
+ dropTableScheduler.start();
recover();
}
@@ -225,9 +234,10 @@ public class ProcedureExecutor<Env> {
waitingTimeoutList.forEach(timeoutExecutor::add);
- failedList.forEach(scheduler::addBack);
- runnableList.forEach(scheduler::addBack);
+ failedList.forEach(procedure ->
getScheduler(procedure).addBack(procedure));
+ runnableList.forEach(procedure ->
getScheduler(procedure).addBack(procedure));
scheduler.signalAll();
+ dropTableScheduler.signalAll();
}
private List<Procedure<Env>> getProcedureListFromDifferentVersion() {
@@ -295,7 +305,7 @@ public class ProcedureExecutor<Env> {
});
}
- public void startWorkers() {
+ public synchronized void startWorkers() {
if (!running.compareAndSet(false, true)) {
LOG.warn(ProcedureMessages.ALREADY_RUNNING);
return;
@@ -305,6 +315,9 @@ public class ProcedureExecutor<Env> {
for (WorkerThread workerThread : workerThreads) {
workerThread.start();
}
+ for (WorkerThread workerThread : dropTableWorkerThreads) {
+ workerThread.start();
+ }
LOG.info(ProcedureMessages.PROCEDURE_WORKERS_ARE_STARTED,
workerThreads.size());
}
@@ -392,7 +405,7 @@ public class ProcedureExecutor<Env> {
break;
case LOCK_YIELD_WAIT:
rootProcStack.unsetRollback();
- scheduler.yield(lockStateResult.procedure);
+
getScheduler(lockStateResult.procedure).yield(lockStateResult.procedure);
break;
default:
throw new UnsupportedOperationException();
@@ -409,7 +422,7 @@ public class ProcedureExecutor<Env> {
lockEventWaitProcedure = proc;
break;
case LOCK_YIELD_WAIT:
- scheduler.yield(proc);
+ getScheduler(proc).yield(proc);
break;
default:
throw new UnsupportedOperationException();
@@ -560,7 +573,7 @@ public class ProcedureExecutor<Env> {
RetryUtils.executeWithEndlessBackoffRetry(
() -> store.update(parent), "count down children procedure");
// do not add this procedure when exception occurred
- scheduler.addFront(parent);
+ getScheduler(parent).addFront(parent);
LOG.info(
ProcedureMessages.LOG_FINISHED_SUBPROCEDURE_PID_ARG_RESUME_PROCESSING_PPID_ARG_93ED990B,
proc.getProcId(),
@@ -577,7 +590,7 @@ public class ProcedureExecutor<Env> {
for (Procedure<Env> subproc : subprocs) {
subproc.updateMetricsOnSubmit(getEnvironment());
procedures.put(subproc.getProcId(), subproc);
- scheduler.addFront(subproc);
+ getScheduler(subproc).addFront(subproc);
LOG.info(ProcedureMessages.SUB_PROCEDURE_PID_HAS_BEEN_SUBMITTED,
subproc.getProcId());
}
}
@@ -665,7 +678,7 @@ public class ProcedureExecutor<Env> {
private void yieldProcedure(Procedure<Env> proc) {
releaseLock(proc, false);
- scheduler.yield(proc);
+ getScheduler(proc).yield(proc);
}
/**
@@ -848,7 +861,7 @@ public class ProcedureExecutor<Env> {
}
rollbackStack.put(currentProcId, stack);
procedures.put(currentProcId, procedure);
- scheduler.addBack(procedure);
+ getScheduler(procedure).addBack(procedure);
return procedure.getProcId();
}
@@ -862,22 +875,25 @@ public class ProcedureExecutor<Env> {
}
private class WorkerThread extends StoppableThread {
+ private final ProcedureScheduler workerScheduler;
private final AtomicLong startTime = new AtomicLong(Long.MAX_VALUE);
private final AtomicReference<Procedure<Env>> activeProcedure = new
AtomicReference<>();
protected long keepAliveTime = -1;
- public WorkerThread(ThreadGroup threadGroup) {
- this(threadGroup, "ProcedureCoreWorker-");
+ public WorkerThread(ThreadGroup threadGroup, ProcedureScheduler
workerScheduler) {
+ this(threadGroup, workerScheduler, "ProcedureCoreWorker-");
}
- public WorkerThread(ThreadGroup threadGroup, String prefix) {
+ public WorkerThread(
+ ThreadGroup threadGroup, ProcedureScheduler workerScheduler, String
prefix) {
super(threadGroup, prefix + workId.incrementAndGet());
+ this.workerScheduler = workerScheduler;
setDaemon(true);
}
@Override
public void sendStopSignal() {
- scheduler.signalAll();
+ workerScheduler.signalAll();
}
@Override
@@ -885,7 +901,7 @@ public class ProcedureExecutor<Env> {
long lastUpdated = System.currentTimeMillis();
try {
while (isRunning() && keepAlive(lastUpdated)) {
- Procedure<Env> procedure = scheduler.poll(keepAliveTime,
TimeUnit.MILLISECONDS);
+ Procedure<Env> procedure = workerScheduler.poll(keepAliveTime,
TimeUnit.MILLISECONDS);
if (procedure == null) {
Thread.sleep(1000);
continue;
@@ -942,7 +958,11 @@ public class ProcedureExecutor<Env> {
} finally {
LOG.info(ProcedureMessages.PROCEDURE_WORKER_TERMINATED, getName());
}
- workerThreads.remove(this);
+ if (workerScheduler == dropTableScheduler) {
+ dropTableWorkerThreads.remove(this);
+ } else {
+ workerThreads.remove(this);
+ }
}
protected boolean keepAlive(long lastUpdated) {
@@ -968,7 +988,7 @@ public class ProcedureExecutor<Env> {
private final class TemporaryWorkerThread extends WorkerThread {
public TemporaryWorkerThread(ThreadGroup group) {
- super(group, "ProcedureTemporaryWorker-");
+ super(group, scheduler, "ProcedureTemporaryWorker-");
this.keepAliveTime = TimeUnit.SECONDS.toMillis(10);
}
@@ -1047,6 +1067,12 @@ public class ProcedureExecutor<Env> {
return workers == null ? 0 : workers.size();
}
+ @TestOnly
+ public int getDropTableWorkerThreadCount() {
+ final CopyOnWriteArrayList<WorkerThread> workers = dropTableWorkerThreads;
+ return workers == null ? 0 : workers.size();
+ }
+
public long getActiveWorkerThreadCount() {
final CopyOnWriteArrayList<WorkerThread> workers = workerThreads;
return workers == null
@@ -1058,12 +1084,13 @@ public class ProcedureExecutor<Env> {
return running.get();
}
- public void stop() {
+ public synchronized void stop() {
if (!running.getAndSet(false)) {
return;
}
LOG.info(ProcedureMessages.STOPPING);
scheduler.stop();
+ dropTableScheduler.stop();
timeoutExecutor.sendStopSignal();
}
@@ -1073,6 +1100,9 @@ public class ProcedureExecutor<Env> {
for (WorkerThread workerThread : workerThreads) {
workerThread.awaitTermination();
}
+ for (WorkerThread workerThread : dropTableWorkerThreads) {
+ workerThread.awaitTermination();
+ }
}
public boolean isStarted(long procId) {
@@ -1091,6 +1121,12 @@ public class ProcedureExecutor<Env> {
return procedures;
}
+ public List<Procedure<Env>> getCompletedProcedures() {
+ final List<Procedure<Env>> result = new ArrayList<>();
+ completed.values().forEach(container ->
result.add(container.getProcedure()));
+ return result;
+ }
+
// -----------------------------CLIENT
IMPLEMENTATION-----------------------------------
/**
* Submit a new root-procedure to the executor, called by client.
@@ -1116,10 +1152,42 @@ public class ProcedureExecutor<Env> {
return pushProcedure(procedure);
}
+ /** Submit only after the procedure has been stored, for requests
acknowledged asynchronously. */
+ public long submitProcedureAndPersist(final Procedure<Env> procedure) throws
Exception {
+ Preconditions.checkArgument(procedure.getState() ==
ProcedureState.INITIALIZING);
+ Preconditions.checkArgument(
+ !procedure.hasParent(),
ProcedureMessages.EXCEPTION_UNEXPECTED_PARENT_444B4289, procedure);
+ procedure.setProcId(store.getNextProcId());
+ procedure.setProcRunnable();
+ store.update(procedure);
+ LOG.debug(ProcedureMessages.IS_STORED, procedure);
+ return pushProcedure(procedure);
+ }
+
public ProcedureScheduler getScheduler() {
return scheduler;
}
+ public ProcedureScheduler getScheduler(final Procedure<?> procedure) {
+ return procedure instanceof DropTableProcedure && !(procedure instanceof
DropViewProcedure)
+ ? getDropTableScheduler()
+ : scheduler;
+ }
+
+ private synchronized ProcedureScheduler getDropTableScheduler() {
+ // Recovery can enqueue drops before startWorkers(), while new submissions
arrive after it.
+ // Share the lifecycle lock so the first drop creates and starts exactly
one worker.
+ if (dropTableWorkerThreads.isEmpty()) {
+ final WorkerThread worker =
+ new WorkerThread(threadGroup, dropTableScheduler,
"DropTableProcedureWorker-");
+ dropTableWorkerThreads.add(worker);
+ if (isRunning()) {
+ worker.start();
+ }
+ }
+ return dropTableScheduler;
+ }
+
public Env getEnvironment() {
return environment;
}
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/TimeoutExecutorThread.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/TimeoutExecutorThread.java
index d8e16ca1b1f..23cfb6c089f 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/TimeoutExecutorThread.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/TimeoutExecutorThread.java
@@ -104,7 +104,7 @@ public class TimeoutExecutorThread<Env> extends
StoppableThread {
// procedure.
LOGGER.warn(ProcedureMessages.FAILED_TO_UPDATE_PROCEDURE,
procedure, e);
}
- executor.getScheduler().addFront(procedure);
+ executor.getScheduler(procedure).addFront(procedure);
}
}
}
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 37f3b85c675..64ea7a04be1 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
@@ -70,6 +70,14 @@ public class DropTableProcedure extends
AbstractAlterOrDropTableProcedure<DropTa
super(database, tableName, queryId, isGeneratedByPipe);
}
+ public String getDropProgress() {
+ if (isFinished()) {
+ return getState().name();
+ }
+ final DropTableState state = getCurrentState();
+ return state == null ? getState().name() : state.name();
+ }
+
@Override
protected String getActionMessage() {
return "drop table";
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java
index b6fe2c0fc21..1f89e39d448 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java
@@ -210,6 +210,7 @@ import
org.apache.iotdb.confignode.rpc.thrift.TShowConfigNodesResp;
import
org.apache.iotdb.confignode.rpc.thrift.TShowDataNodes4InformationSchemaResp;
import org.apache.iotdb.confignode.rpc.thrift.TShowDataNodesResp;
import org.apache.iotdb.confignode.rpc.thrift.TShowDatabaseResp;
+import org.apache.iotdb.confignode.rpc.thrift.TShowDropTableProceduresResp;
import org.apache.iotdb.confignode.rpc.thrift.TShowPipePluginReq;
import org.apache.iotdb.confignode.rpc.thrift.TShowPipeReq;
import org.apache.iotdb.confignode.rpc.thrift.TShowPipeResp;
@@ -1498,6 +1499,11 @@ public class ConfigNodeRPCServiceProcessor implements
IConfigNodeRPCService.Ifac
return configManager.showTables4InformationSchema();
}
+ @Override
+ public TShowDropTableProceduresResp showDropTableProcedures() {
+ return configManager.showDropTableProcedures();
+ }
+
@Override
public TDescTableResp describeTable(
final String database, final String tableName, final boolean isDetails) {
diff --git
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/client/async/CnToDnInternalServiceAsyncRequestManagerTest.java
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/client/async/CnToDnInternalServiceAsyncRequestManagerTest.java
new file mode 100644
index 00000000000..3769782bf12
--- /dev/null
+++
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/client/async/CnToDnInternalServiceAsyncRequestManagerTest.java
@@ -0,0 +1,114 @@
+/*
+ * 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;
+
+import org.apache.iotdb.common.rpc.thrift.TEndPoint;
+import org.apache.iotdb.commons.client.ClientManager;
+import
org.apache.iotdb.commons.client.async.AsyncDataNodeInternalServiceClient;
+import org.apache.iotdb.commons.client.property.ThriftClientProperty;
+
+import org.apache.thrift.TException;
+import org.apache.thrift.async.TAsyncClientManager;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+public class CnToDnInternalServiceAsyncRequestManagerTest {
+
+ private static final int DEFAULT_TIMEOUT_MS = 1_000;
+
+ private final CnToDnInternalServiceAsyncRequestManager requestManager =
+ CnToDnInternalServiceAsyncRequestManager.getInstance();
+ private final TEndPoint endpoint = new TEndPoint("127.0.0.1", 9000);
+ private ClientManager<TEndPoint, AsyncDataNodeInternalServiceClient>
clientManager;
+ private AsyncDataNodeInternalServiceClient client;
+
+ @Before
+ @SuppressWarnings("unchecked")
+ public void setUp() throws Exception {
+ clientManager = mock(ClientManager.class);
+ client =
+ new AsyncDataNodeInternalServiceClient(
+ new ThriftClientProperty.Builder()
+ .setConnectionTimeoutMs(DEFAULT_TIMEOUT_MS)
+ .setPrintLogWhenEncounterException(false)
+ .build(),
+ endpoint,
+ mock(TAsyncClientManager.class),
+ clientManager);
+ }
+
+ @After
+ public void tearDown() {
+ client.invalidate();
+ }
+
+ @Test
+ public void
dropTableDeletionDisablesDeadlineAndRestoresTimeoutOnCompletion() {
+ for (final CnToDnAsyncRequestType requestType :
+ new CnToDnAsyncRequestType[] {
+ CnToDnAsyncRequestType.DELETE_DATA_FOR_DROP_TABLE,
+ CnToDnAsyncRequestType.DELETE_DEVICES_FOR_DROP_TABLE
+ }) {
+ requestManager.adjustClientTimeoutIfNecessary(requestType, client, null);
+ assertFalse(client.hasTimeout());
+
+ client.onComplete();
+ assertTrue(client.hasTimeout());
+ assertEquals(DEFAULT_TIMEOUT_MS, client.getTimeout());
+ }
+ verify(clientManager, times(2)).returnClient(endpoint, client);
+ }
+
+ @Test
+ public void dropTableDeletionRestoresTimeoutOnError() {
+ requestManager.adjustClientTimeoutIfNecessary(
+ CnToDnAsyncRequestType.DELETE_DATA_FOR_DROP_TABLE, client, null);
+ assertFalse(client.hasTimeout());
+
+ client.onError(new TException("Deletion failed"));
+ assertTrue(client.hasTimeout());
+ assertEquals(DEFAULT_TIMEOUT_MS, client.getTimeout());
+ verify(clientManager).returnClient(endpoint, client);
+ }
+
+ @Test
+ public void cacheInvalidationKeepsDefaultAndExplicitTimeouts() {
+ requestManager.adjustClientTimeoutIfNecessary(
+ CnToDnAsyncRequestType.PRE_DELETE_TABLE, client, null);
+ assertTrue(client.hasTimeout());
+ assertEquals(DEFAULT_TIMEOUT_MS, client.getTimeout());
+
+ requestManager.adjustClientTimeoutIfNecessary(
+ CnToDnAsyncRequestType.PRE_DELETE_TABLE, client, 5_000L);
+ assertTrue(client.hasTimeout());
+ assertEquals(5_000L, client.getTimeout());
+
+ client.onComplete();
+ assertEquals(DEFAULT_TIMEOUT_MS, client.getTimeout());
+ }
+}
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 7da83c02a9d..fb141aab270 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
@@ -20,7 +20,14 @@
package org.apache.iotdb.confignode.procedure.impl.schema.table;
import org.apache.iotdb.commons.exception.IllegalPathException;
+import org.apache.iotdb.confignode.procedure.NoopProcedureStore;
+import org.apache.iotdb.confignode.procedure.Procedure;
+import org.apache.iotdb.confignode.procedure.ProcedureExecutor;
+import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv;
+import
org.apache.iotdb.confignode.procedure.impl.schema.table.view.DropViewProcedure;
+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.junit.Assert;
import org.junit.Test;
@@ -29,8 +36,178 @@ import java.io.ByteArrayOutputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
public class DropTableProcedureTest {
+ @Test
+ public void dropTableDoesNotOccupyRegularProcedureWorker() throws Exception {
+ final NoopProcedureStore store = new NoopProcedureStore();
+ final ProcedureExecutor<ConfigNodeProcedureEnv> executor = new
ProcedureExecutor<>(null, store);
+ final CountDownLatch dropStarted = new CountDownLatch(1);
+ final CountDownLatch releaseDrop = new CountDownLatch(1);
+ final CountDownLatch regularCompleted = new CountDownLatch(1);
+ Assert.assertEquals(0, executor.getDropTableWorkerThreadCount());
+ store.start();
+ executor.init(1);
+ Assert.assertEquals(0, executor.getDropTableWorkerThreadCount());
+ executor.startWorkers();
+ try {
+ Assert.assertSame(
+ executor.getScheduler(), executor.getScheduler(new
DropViewProcedure(false)));
+ Assert.assertEquals(0, executor.getDropTableWorkerThreadCount());
+ final DropTableProcedure dropProcedure =
+ new DropTableProcedure("db", "table", "drop", false) {
+ @Override
+ protected Flow executeFromState(
+ final ConfigNodeProcedureEnv env, final DropTableState state)
+ throws InterruptedException {
+ dropStarted.countDown();
+ releaseDrop.await();
+ return Flow.NO_MORE_STATE;
+ }
+ };
+ executor.submitProcedureAndPersist(dropProcedure);
+ Assert.assertTrue(dropStarted.await(10, TimeUnit.SECONDS));
+ Assert.assertEquals(1, executor.getDropTableWorkerThreadCount());
+ Assert.assertEquals(
+ DropTableState.CHECK_AND_INVALIDATE_TABLE.name(),
dropProcedure.getDropProgress());
+
+ executor.submitProcedure(
+ new Procedure<ConfigNodeProcedureEnv>() {
+ @Override
+ protected Procedure<ConfigNodeProcedureEnv>[] execute(
+ final ConfigNodeProcedureEnv env) {
+ regularCompleted.countDown();
+ return null;
+ }
+
+ @Override
+ protected void rollback(final ConfigNodeProcedureEnv env) {}
+ });
+ Assert.assertTrue(regularCompleted.await(10, TimeUnit.SECONDS));
+ } finally {
+ releaseDrop.countDown();
+ executor.stop();
+ executor.join();
+ store.stop();
+ }
+ Assert.assertEquals(0, executor.getDropTableWorkerThreadCount());
+ }
+
+ @Test
+ public void recoveredDropStartsWorkerAndCompletedDropDoesNot() throws
Exception {
+ final CountDownLatch dropStarted = new CountDownLatch(1);
+ final CountDownLatch releaseDrop = new CountDownLatch(1);
+ final DropTableProcedure dropProcedure =
+ new DropTableProcedure("db", "table", "drop", false) {
+ @Override
+ protected Flow executeFromState(
+ final ConfigNodeProcedureEnv env, final DropTableState state)
+ throws InterruptedException {
+ dropStarted.countDown();
+ releaseDrop.await();
+ return Flow.NO_MORE_STATE;
+ }
+ };
+ dropProcedure.setProcId(0);
+ dropProcedure.setProcRunnable();
+ final NoopProcedureStore store =
+ new NoopProcedureStore() {
+ @Override
+ public List<Procedure> load() {
+ return Collections.singletonList(dropProcedure);
+ }
+ };
+ final ProcedureExecutor<ConfigNodeProcedureEnv> executor = new
ProcedureExecutor<>(null, store);
+ store.start();
+ executor.init(1);
+ try {
+ Assert.assertEquals(1, executor.getDropTableWorkerThreadCount());
+ Assert.assertEquals(1, dropStarted.getCount());
+ executor.startWorkers();
+ Assert.assertTrue(dropStarted.await(10, TimeUnit.SECONDS));
+ releaseDrop.countDown();
+ ProcedureTestUtil.waitForProcedure(executor, dropProcedure.getProcId());
+ Assert.assertTrue(dropProcedure.isSuccess());
+ executor.stop();
+ executor.join();
+
+ executor.init(1);
+ executor.startWorkers();
+ Assert.assertEquals(0, executor.getDropTableWorkerThreadCount());
+ Assert.assertEquals(
+ Collections.singletonList(dropProcedure),
executor.getCompletedProcedures());
+ } finally {
+ releaseDrop.countDown();
+ executor.stop();
+ executor.join();
+ store.stop();
+ }
+ }
+
+ @Test
+ public void concurrentFirstDropsShareOneWorker() throws Exception {
+ final NoopProcedureStore store = new NoopProcedureStore();
+ final ProcedureExecutor<ConfigNodeProcedureEnv> executor = new
ProcedureExecutor<>(null, store);
+ final ExecutorService submitters = Executors.newFixedThreadPool(4);
+ final CountDownLatch submittersReady = new CountDownLatch(4);
+ final CountDownLatch startSubmissions = new CountDownLatch(1);
+ final CountDownLatch dropStarted = new CountDownLatch(1);
+ final CountDownLatch releaseDrops = new CountDownLatch(1);
+ store.start();
+ executor.init(1);
+ executor.startWorkers();
+ try {
+ final List<Future<Long>> submissions = new ArrayList<>();
+ for (int i = 0; i < 4; i++) {
+ final DropTableProcedure dropProcedure =
+ new DropTableProcedure("db", "table" + i, "drop" + i, false) {
+ @Override
+ protected Flow executeFromState(
+ final ConfigNodeProcedureEnv env, final DropTableState state)
+ throws InterruptedException {
+ dropStarted.countDown();
+ releaseDrops.await();
+ return Flow.NO_MORE_STATE;
+ }
+ };
+ submissions.add(
+ submitters.submit(
+ () -> {
+ submittersReady.countDown();
+ startSubmissions.await();
+ return executor.submitProcedureAndPersist(dropProcedure);
+ }));
+ }
+ Assert.assertTrue(submittersReady.await(10, TimeUnit.SECONDS));
+ startSubmissions.countDown();
+ for (final Future<Long> submission : submissions) {
+ submission.get(10, TimeUnit.SECONDS);
+ }
+ Assert.assertTrue(dropStarted.await(10, TimeUnit.SECONDS));
+ Assert.assertEquals(1, executor.getDropTableWorkerThreadCount());
+ releaseDrops.countDown();
+ for (final Future<Long> submission : submissions) {
+ ProcedureTestUtil.waitForProcedure(executor, submission.get());
+ }
+ Assert.assertEquals(4, executor.getCompletedProcedures().size());
+ } finally {
+ startSubmissions.countDown();
+ releaseDrops.countDown();
+ submitters.shutdownNow();
+ executor.stop();
+ executor.join();
+ store.stop();
+ }
+ }
+
@Test
public void serializeDeserializeTest() throws IllegalPathException,
IOException {
final DropTableProcedure dropTableProcedure =
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/client/ConfigNodeClient.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/client/ConfigNodeClient.java
index a9945869fbf..9f1e90fda10 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/client/ConfigNodeClient.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/client/ConfigNodeClient.java
@@ -164,6 +164,7 @@ import
org.apache.iotdb.confignode.rpc.thrift.TShowConfigNodesResp;
import
org.apache.iotdb.confignode.rpc.thrift.TShowDataNodes4InformationSchemaResp;
import org.apache.iotdb.confignode.rpc.thrift.TShowDataNodesResp;
import org.apache.iotdb.confignode.rpc.thrift.TShowDatabaseResp;
+import org.apache.iotdb.confignode.rpc.thrift.TShowDropTableProceduresResp;
import org.apache.iotdb.confignode.rpc.thrift.TShowPipePluginReq;
import org.apache.iotdb.confignode.rpc.thrift.TShowPipeReq;
import org.apache.iotdb.confignode.rpc.thrift.TShowPipeResp;
@@ -1274,6 +1275,12 @@ public class ConfigNodeClient extends
AbstractConfigNodeClient<ConfigNodeClient>
() -> client.showTables4InformationSchema(), resp ->
!updateConfigNodeLeader(resp.status));
}
+ @Override
+ public TShowDropTableProceduresResp showDropTableProcedures() throws
TException {
+ return executeRemoteCallWithRetry(
+ () -> client.showDropTableProcedures(), resp ->
!updateConfigNodeLeader(resp.status));
+ }
+
@Override
public TDescTableResp describeTable(
final String database, final String tableName, final boolean isDetails)
throws TException {
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 f57fc00154e..5b92f259373 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
@@ -59,11 +59,13 @@ import
org.apache.iotdb.confignode.rpc.thrift.TConfigNodeInfo4InformationSchema;
import org.apache.iotdb.confignode.rpc.thrift.TDataNodeInfo4InformationSchema;
import org.apache.iotdb.confignode.rpc.thrift.TDatabaseInfo;
import org.apache.iotdb.confignode.rpc.thrift.TDescTable4InformationSchemaResp;
+import org.apache.iotdb.confignode.rpc.thrift.TDropTableProcedureInfo;
import org.apache.iotdb.confignode.rpc.thrift.TGetDatabaseReq;
import org.apache.iotdb.confignode.rpc.thrift.TGetUdfTableReq;
import org.apache.iotdb.confignode.rpc.thrift.TNodeVersionInfo;
import org.apache.iotdb.confignode.rpc.thrift.TRegionInfo;
import org.apache.iotdb.confignode.rpc.thrift.TShowClusterResp;
+import org.apache.iotdb.confignode.rpc.thrift.TShowDropTableProceduresResp;
import org.apache.iotdb.confignode.rpc.thrift.TShowPipeInfo;
import org.apache.iotdb.confignode.rpc.thrift.TShowPipeReq;
import org.apache.iotdb.confignode.rpc.thrift.TShowRegionReq;
@@ -193,6 +195,8 @@ public class InformationSchemaContentSupplierFactory {
return new DatabaseSupplier(dataTypes, userEntity);
case InformationSchema.TABLES:
return new TableSupplier(dataTypes, userEntity);
+ case InformationSchema.DROP_TABLE_PROCEDURES:
+ return new DropTableProceduresSupplier(dataTypes, userEntity);
case InformationSchema.COLUMNS:
return new ColumnSupplier(dataTypes, userEntity);
case InformationSchema.REGIONS:
@@ -486,6 +490,70 @@ public class InformationSchemaContentSupplierFactory {
}
}
+ private static class DropTableProceduresSupplier extends TsBlockSupplier {
+ private final Iterator<TDropTableProcedureInfo> procedures;
+ private final UserEntity userEntity;
+ private TDropTableProcedureInfo currentProcedure;
+
+ private DropTableProceduresSupplier(
+ final List<TSDataType> dataTypes, final UserEntity userEntity) throws
Exception {
+ super(dataTypes);
+ this.userEntity = userEntity;
+ try (final ConfigNodeClient client =
+
ConfigNodeClientManager.getInstance().borrowClient(ConfigNodeInfo.CONFIG_REGION_ID))
{
+ final TShowDropTableProceduresResp response =
client.showDropTableProcedures();
+ if (response.getStatus().getCode() !=
TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ throw new IoTDBRuntimeException(response.getStatus());
+ }
+ final List<TDropTableProcedureInfo> result =
response.getDropTableProcedureInfoList();
+ procedures =
+ (result == null ? Collections.<TDropTableProcedureInfo>emptyList()
: result).iterator();
+ }
+ }
+
+ @Override
+ protected void constructLine() {
+ columnBuilders[0].writeBinary(
+ new Binary(currentProcedure.getDatabase(),
TSFileConfig.STRING_CHARSET));
+ columnBuilders[1].writeBinary(
+ new Binary(currentProcedure.getTableName(),
TSFileConfig.STRING_CHARSET));
+ columnBuilders[2].writeLong(currentProcedure.getProcedureId());
+ columnBuilders[3].writeBinary(
+ new Binary(currentProcedure.getState(),
TSFileConfig.STRING_CHARSET));
+ columnBuilders[4].writeBinary(
+ new Binary(currentProcedure.getProgress(),
TSFileConfig.STRING_CHARSET));
+ if (currentProcedure.isSetErrorMessage()) {
+ columnBuilders[5].writeBinary(
+ new Binary(currentProcedure.getErrorMessage(),
TSFileConfig.STRING_CHARSET));
+ } else {
+ columnBuilders[5].appendNull();
+ }
+ resultBuilder.declarePosition();
+ currentProcedure = null;
+ }
+
+ @Override
+ public boolean hasNext() {
+ if (currentProcedure != null) {
+ return true;
+ }
+ while (procedures.hasNext()) {
+ final TDropTableProcedureInfo procedure = procedures.next();
+ if (canShowDB(accessControl, userEntity.getUsername(),
procedure.getDatabase(), userEntity)
+ && canShowTable(
+ accessControl,
+ userEntity.getUsername(),
+ procedure.getDatabase(),
+ procedure.getTableName(),
+ userEntity)) {
+ currentProcedure = procedure;
+ return true;
+ }
+ }
+ return false;
+ }
+ }
+
public static class ColumnSupplier extends TsBlockSupplier {
private final Iterator<Map.Entry<String, Map<String,
TableColumnDetailInfo>>> dbIterator;
private Iterator<Map.Entry<String, TableColumnDetailInfo>>
tableInfoIterator;
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/DataNodeLocationSupplierFactory.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/DataNodeLocationSupplierFactory.java
index b4d8e8d7a43..fdb6d2c26bf 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/DataNodeLocationSupplierFactory.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/DataNodeLocationSupplierFactory.java
@@ -146,6 +146,7 @@ public class DataNodeLocationSupplierFactory {
return getReadableDataNodeLocations();
case InformationSchema.DATABASES:
case InformationSchema.TABLES:
+ case InformationSchema.DROP_TABLE_PROCEDURES:
case InformationSchema.COLUMNS:
case InformationSchema.REGIONS:
case InformationSchema.PIPES:
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/AbstractConfigNodeClient.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/AbstractConfigNodeClient.java
index a95bf954888..86bc416653e 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/AbstractConfigNodeClient.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/AbstractConfigNodeClient.java
@@ -162,6 +162,7 @@ import
org.apache.iotdb.confignode.rpc.thrift.TShowConfigNodesResp;
import
org.apache.iotdb.confignode.rpc.thrift.TShowDataNodes4InformationSchemaResp;
import org.apache.iotdb.confignode.rpc.thrift.TShowDataNodesResp;
import org.apache.iotdb.confignode.rpc.thrift.TShowDatabaseResp;
+import org.apache.iotdb.confignode.rpc.thrift.TShowDropTableProceduresResp;
import org.apache.iotdb.confignode.rpc.thrift.TShowPipePluginReq;
import org.apache.iotdb.confignode.rpc.thrift.TShowPipeReq;
import org.apache.iotdb.confignode.rpc.thrift.TShowPipeResp;
@@ -1302,6 +1303,11 @@ public abstract class AbstractConfigNodeClient<C extends
AbstractConfigNodeClien
return null;
}
+ @Override
+ public TShowDropTableProceduresResp showDropTableProcedures() throws
TException {
+ return null;
+ }
+
@Override
public TDescTableResp describeTable(String database, String tableName,
boolean isDetails)
throws TException {
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 239478b13fe..03ba898f77f 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
@@ -305,6 +305,9 @@ public class ColumnHeaderConstant {
public static final String PIPE_PROCESSOR_TABLE_MODEL = "pipe_processor";
public static final String PIPE_SINK_TABLE_MODEL = "pipe_sink";
public static final String EXCEPTION_MESSAGE_TABLE_MODEL =
"exception_message";
+ 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 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 00ef05c5afa..9f3bf0bb2df 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
@@ -47,6 +47,7 @@ public class InformationSchema {
public static final String QUERIES = "queries";
public static final String DATABASES = "databases";
public static final String TABLES = "tables";
+ public static final String DROP_TABLE_PROCEDURES = "drop_table_procedures";
public static final String COLUMNS = "columns";
public static final String REGIONS = "regions";
public static final String PIPES = "pipes";
@@ -148,6 +149,25 @@ public class InformationSchema {
tableTable.removeColumnSchema(TsTable.TIME_COLUMN_NAME);
schemaTables.put(TABLES, tableTable);
+ final TsTable dropTableProceduresTable = new
TsTable(DROP_TABLE_PROCEDURES);
+ dropTableProceduresTable.addColumnSchema(
+ new TagColumnSchema(
+ ColumnHeaderConstant.DATABASE.toLowerCase(Locale.ENGLISH),
TSDataType.STRING));
+ dropTableProceduresTable.addColumnSchema(
+ new TagColumnSchema(ColumnHeaderConstant.TABLE_NAME_TABLE_MODEL,
TSDataType.STRING));
+ dropTableProceduresTable.addColumnSchema(
+ new
AttributeColumnSchema(ColumnHeaderConstant.PROCEDURE_ID_TABLE_MODEL,
TSDataType.INT64));
+ dropTableProceduresTable.addColumnSchema(
+ new AttributeColumnSchema(
+ ColumnHeaderConstant.STATE.toLowerCase(Locale.ENGLISH),
TSDataType.STRING));
+ dropTableProceduresTable.addColumnSchema(
+ new AttributeColumnSchema(ColumnHeaderConstant.PROGRESS_TABLE_MODEL,
TSDataType.STRING));
+ dropTableProceduresTable.addColumnSchema(
+ new AttributeColumnSchema(
+ ColumnHeaderConstant.ERROR_MESSAGE_TABLE_MODEL,
TSDataType.STRING));
+ dropTableProceduresTable.removeColumnSchema(TsTable.TIME_COLUMN_NAME);
+ schemaTables.put(DROP_TABLE_PROCEDURES, dropTableProceduresTable);
+
final TsTable columnTable = new TsTable(COLUMNS);
columnTable.addColumnSchema(
new TagColumnSchema(
diff --git a/iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift
b/iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift
index 48921651ee2..e2f2595fa7d 100644
--- a/iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift
+++ b/iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift
@@ -1314,6 +1314,20 @@ struct TShowTable4InformationSchemaResp {
2: optional map<string, list<TTableInfo>> databaseTableInfoMap
}
+struct TShowDropTableProceduresResp {
+ 1: required common.TSStatus status
+ 2: optional list<TDropTableProcedureInfo> dropTableProcedureInfoList
+}
+
+struct TDropTableProcedureInfo {
+ 1: required string database
+ 2: required string tableName
+ 3: required i64 procedureId
+ 4: required string state
+ 5: required string progress
+ 6: optional string errorMessage
+}
+
struct TDescTableResp {
1: required common.TSStatus status
2: optional binary tableInfo
@@ -2142,6 +2156,7 @@ service IConfigNodeRPCService {
TShowTableResp showTables(string database, bool isDetails)
TShowTable4InformationSchemaResp showTables4InformationSchema()
+ TShowDropTableProceduresResp showDropTableProcedures()
TDescTableResp describeTable(string database, string tableName, bool
isDetails)
