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 f87326459262f..c6063a2c59d84 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.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 void testDeleteTable() throws SQLException { 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 c7f85ea500ea7..c84f197a6d897 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 void testInformationSchema() throws SQLException { "current_queries,INF,", "data_nodes,INF,", "databases,INF,", + "drop_table_procedures,INF,", "functions,INF,", "keywords,INF,", "nodes,INF,", @@ -828,6 +829,7 @@ public void testInformationSchema() throws SQLException { 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,", @@ -853,7 +855,7 @@ public void testInformationSchema() throws SQLException { 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 4c13f660972eb..4589ec5d49ae7 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 void testManageTable() { Collections.singleton("null,")); statement.execute("drop table table2"); + awaitDropProcedure(statement, "test2", "table2"); try { statement.executeQuery("describe table2"); fail(); @@ -1229,4 +1230,69 @@ public void testTreeViewTable() throws Exception { } } } + + @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 7ca698afa2c93..e73573ea0e8a0 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 @@ protected AsyncRequestRPCHandler b @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 2a6b7e0f68cfb..680fa87a8d546 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.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 TShowTable4InformationSchemaResp showTables4InformationSchema() { : 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 bae45596812e1..d8241f70e2e0f 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.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; @@ -916,6 +917,8 @@ TPermissionInfoResp login( 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 c97654b649e62..dac83821ba6e8 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.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; @@ -59,6 +61,7 @@ import org.apache.iotdb.confignode.consensus.request.write.region.CreateRegionGroupsPlan; import org.apache.iotdb.confignode.i18n.ManagerMessages; 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; @@ -158,6 +161,7 @@ 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; @@ -2376,15 +2380,63 @@ public TSStatus alterTableColumnDataType(TAlterOrDropTableReq req) { 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 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> 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> 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 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) { @@ -2484,6 +2536,35 @@ public Map> getAllExecutingTables() { return result; } + /** Returns running drops and completed results retained by the executor's shared TTL cleaner. */ + public List getDropTableProcedures() { + final Map 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 b5f9a021426b1..075c5c587a2fe 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.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 { // Metrics may be scraped before init() and concurrently with initialization during a ConfigNode // leader transition. private volatile CopyOnWriteArrayList workerThreads; + private volatile CopyOnWriteArrayList dropTableWorkerThreads; private TimeoutExecutorThread timeoutExecutor; @@ -89,6 +92,9 @@ public class ProcedureExecutor { private CompletedProcedureRecycler 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 static boolean isProcedureExecutionThread() { 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 void init(int numThreads) { 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 @@ private void recover() { 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> getProcedureListFromDifferentVersion() { @@ -295,7 +305,7 @@ private void restoreLocks() { }); } - public void startWorkers() { + public synchronized void startWorkers() { if (!running.compareAndSet(false, true)) { LOG.warn(ProcedureMessages.ALREADY_RUNNING); return; @@ -305,6 +315,9 @@ public void startWorkers() { 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 @@ private void executeProcedure(Procedure proc) { 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 @@ private void executeProcedure(Procedure proc) { lockEventWaitProcedure = proc; break; case LOCK_YIELD_WAIT: - scheduler.yield(proc); + getScheduler(proc).yield(proc); break; default: throw new UnsupportedOperationException(); @@ -560,7 +573,7 @@ private void countDownChildren(RootProcedureStack rootProcStack, Procedure 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 @@ private void submitChildrenProcedures(Procedure[] subprocs) { for (Procedure 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 @@ private Procedure[] initializeChildren( private void yieldProcedure(Procedure proc) { releaseLock(proc, false); - scheduler.yield(proc); + getScheduler(proc).yield(proc); } /** @@ -833,7 +846,7 @@ private long pushProcedure(Procedure procedure) { } rollbackStack.put(currentProcId, stack); procedures.put(currentProcId, procedure); - scheduler.addBack(procedure); + getScheduler(procedure).addBack(procedure); return procedure.getProcId(); } @@ -847,22 +860,25 @@ private boolean initializeRollbackStackForFailedProcedure( } private class WorkerThread extends StoppableThread { + private final ProcedureScheduler workerScheduler; private final AtomicLong startTime = new AtomicLong(Long.MAX_VALUE); private final AtomicReference> 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 @@ -870,7 +886,7 @@ public void run() { long lastUpdated = System.currentTimeMillis(); try { while (isRunning() && keepAlive(lastUpdated)) { - Procedure procedure = scheduler.poll(keepAliveTime, TimeUnit.MILLISECONDS); + Procedure procedure = workerScheduler.poll(keepAliveTime, TimeUnit.MILLISECONDS); if (procedure == null) { Thread.sleep(1000); continue; @@ -927,7 +943,11 @@ public void run() { } 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) { @@ -953,7 +973,7 @@ public long getCurrentRunTime() { private final class TemporaryWorkerThread extends WorkerThread { public TemporaryWorkerThread(ThreadGroup group) { - super(group, "ProcedureTemporaryWorker-"); + super(group, scheduler, "ProcedureTemporaryWorker-"); this.keepAliveTime = TimeUnit.SECONDS.toMillis(10); } @@ -1032,6 +1052,12 @@ public int getWorkerThreadCount() { return workers == null ? 0 : workers.size(); } + @TestOnly + public int getDropTableWorkerThreadCount() { + final CopyOnWriteArrayList workers = dropTableWorkerThreads; + return workers == null ? 0 : workers.size(); + } + public long getActiveWorkerThreadCount() { final CopyOnWriteArrayList workers = workerThreads; return workers == null @@ -1043,12 +1069,13 @@ public boolean isRunning() { 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(); } @@ -1058,6 +1085,9 @@ public void join() { for (WorkerThread workerThread : workerThreads) { workerThread.awaitTermination(); } + for (WorkerThread workerThread : dropTableWorkerThreads) { + workerThread.awaitTermination(); + } } public boolean isStarted(long procId) { @@ -1076,6 +1106,12 @@ public ConcurrentHashMap> getProcedures() { return procedures; } + public List> getCompletedProcedures() { + final List> 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. @@ -1101,10 +1137,42 @@ public long submitProcedure(Procedure procedure) { return pushProcedure(procedure); } + /** Submit only after the procedure has been stored, for requests acknowledged asynchronously. */ + public long submitProcedureAndPersist(final Procedure 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 d8e16ca1b1f9e..23cfb6c089fef 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 void run() { // 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 37f3b85c675c4..64ea7a04be120 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 DropTableProcedure( 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 b6fe2c0fc2194..1f89e39d44883 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.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 TShowTable4InformationSchemaResp showTables4InformationSchema() { 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 0000000000000..3769782bf1224 --- /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 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 7da83c02a9d83..fb141aab270db 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.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 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() { + @Override + protected Procedure[] 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 load() { + return Collections.singletonList(dropProcedure); + } + }; + final ProcedureExecutor 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 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> 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 submission : submissions) { + submission.get(10, TimeUnit.SECONDS); + } + Assert.assertTrue(dropStarted.await(10, TimeUnit.SECONDS)); + Assert.assertEquals(1, executor.getDropTableWorkerThreadCount()); + releaseDrops.countDown(); + for (final Future 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 a9945869fbf59..9f1e90fda10af 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.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 TShowTable4InformationSchemaResp showTables4InformationSchema() throws TE () -> 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 f4fb59f7ca7a5..2055757a29d13 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.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 static IInformationSchemaContentSupplier getSupplier( 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 boolean hasNext() { } } + private static class DropTableProceduresSupplier extends TsBlockSupplier { + private final Iterator procedures; + private final UserEntity userEntity; + private TDropTableProcedureInfo currentProcedure; + + private DropTableProceduresSupplier( + final List 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 result = response.getDropTableProcedureInfoList(); + procedures = + (result == null ? Collections.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>> dbIterator; private Iterator> 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 b4d8e8d7a4371..fdb6d2c26bfbb 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 List getDataNodeLocations(final String tableName) { 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 a95bf95488878..86bc416653e26 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.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 TShowTable4InformationSchemaResp showTables4InformationSchema() throws TE 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 e4c365818e842..3afcb56d0e7bf 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 @@ -304,6 +304,9 @@ private 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 92f8950a386c6..a390b43d01202 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 512e2e05f05fc..02825b0fb681f 100644 --- a/iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift +++ b/iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift @@ -1305,6 +1305,20 @@ struct TShowTable4InformationSchemaResp { 2: optional map> databaseTableInfoMap } +struct TShowDropTableProceduresResp { + 1: required common.TSStatus status + 2: optional list 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 @@ -2129,6 +2143,7 @@ service IConfigNodeRPCService { TShowTableResp showTables(string database, bool isDetails) TShowTable4InformationSchemaResp showTables4InformationSchema() + TShowDropTableProceduresResp showDropTableProcedures() TDescTableResp describeTable(string database, string tableName, bool isDetails)