[Feature] Make DROP TABLE asynchronous and expose deletion progress

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) {

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'"),

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");
+  }
 }

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) {

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);
 

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<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) {
@@ -2484,6 +2536,35 @@ public Map<String, List<String>> getAllExecutingTables() {
     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,