From 329867e9f67c3768929013022996f9675e8f81fb Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Fri, 16 Oct 2026 09:51:54 +0800 Subject: [PATCH 1/6] [Test] Query drop table procedure progress on a single DataNode --- .../relational/it/schema/IoTDBTableIT.java | 24 +++++++++++++++---- 1 file changed, 19 insertions(+), 5 deletions(-) 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 34bd61d03d52..ef59faff72c6 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 @@ -26,6 +26,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.runtime.ClusterTestConnection; import org.apache.iotdb.rpc.IoTDBConnectionException; import org.apache.iotdb.rpc.StatementExecutionException; @@ -83,7 +84,8 @@ public static void tearDown() throws Exception { public void testManageTable() { try (final Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT); - final Statement statement = connection.createStatement()) { + final Statement statement = connection.createStatement(); + final Statement procedureStatement = createProcedureStatement(connection)) { statement.execute("create database test1"); statement.execute("create database test2 with (ttl=3000000)"); @@ -546,7 +548,7 @@ public void testManageTable() { Collections.singleton("null,")); statement.execute("drop table table2"); - awaitDropProcedure(statement, "test2", "table2"); + awaitDropProcedure(procedureStatement, "test2", "table2"); try { statement.executeQuery("describe table2"); fail(); @@ -1262,7 +1264,8 @@ public void testTreeViewTable() throws Exception { public void testAsyncDropTableProgress() throws Exception { try (final Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT); - final Statement statement = connection.createStatement()) { + final Statement statement = connection.createStatement(); + final Statement procedureStatement = createProcedureStatement(connection)) { statement.execute("create database async_drop_db"); statement.execute("use async_drop_db"); statement.execute("create table drop_target (device string tag, reading int32)"); @@ -1270,13 +1273,13 @@ public void testAsyncDropTableProgress() throws Exception { // 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"); + awaitDropProcedure(procedureStatement, "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( + procedureStatement.executeQuery( "select procedure_id, state, progress, error_message, " + "written_mod_file_count, remaining_mod_file_count " + "from information_schema.drop_table_procedures " @@ -1295,6 +1298,17 @@ public void testAsyncDropTableProgress() throws Exception { } } + private static Statement createProcedureStatement(final Connection connection) + throws SQLException { + // Procedure progress can change between queries on different DataNodes. Query a single node + // without the cluster connection's cross-node result verification. + final Connection procedureConnection = + connection instanceof ClusterTestConnection + ? ((ClusterTestConnection) connection).writeConnection.getUnderlyingConnection() + : connection; + return procedureConnection.createStatement(); + } + private static void awaitDropProcedure( final Statement statement, final String database, final String tableName) throws SQLException { From ff12d5dde3b64b80beefc6f80ac082dc8de18889 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Fri, 16 Oct 2026 10:05:36 +0800 Subject: [PATCH 2/6] [Test] Fix other dynamic status queries and pipe deletion polling --- .../apache/iotdb/db/it/utils/TestUtils.java | 16 ++++ .../manual/enhanced/IoTDBPipeAutoDropIT.java | 66 +++++++-------- .../auto/enhanced/IoTDBPipeAutoDropIT.java | 84 +++++++------------ .../it/db/it/IoTDBDeletionTableIT.java | 8 +- .../relational/it/schema/IoTDBDatabaseIT.java | 7 +- .../schema/IoTDBDropTableModProgressIT.java | 5 +- .../relational/it/schema/IoTDBTableIT.java | 16 +--- 7 files changed, 87 insertions(+), 115 deletions(-) diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/utils/TestUtils.java b/integration-test/src/test/java/org/apache/iotdb/db/it/utils/TestUtils.java index bf1cb4376bb9..aabce4f92f89 100644 --- a/integration-test/src/test/java/org/apache/iotdb/db/it/utils/TestUtils.java +++ b/integration-test/src/test/java/org/apache/iotdb/db/it/utils/TestUtils.java @@ -26,6 +26,7 @@ import org.apache.iotdb.it.env.cluster.env.AbstractEnv; import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper; import org.apache.iotdb.itbase.env.BaseEnv; +import org.apache.iotdb.itbase.runtime.ClusterTestConnection; import org.apache.iotdb.rpc.IoTDBConnectionException; import org.apache.iotdb.rpc.RpcUtils; import org.apache.iotdb.rpc.StatementExecutionException; @@ -83,6 +84,21 @@ public class TestUtils { public static final String TIME_PRECISION_IN_NS = "ns"; + /** + * Creates a statement on the write node without comparing results across DataNodes. + * + *

Use this for node-local diagnostics or changing progress snapshots. Closing the statement + * does not close the supplied connection. + */ + public static Statement createSingleNodeStatement(final Connection connection) + throws SQLException { + final Connection singleNodeConnection = + connection instanceof ClusterTestConnection + ? ((ClusterTestConnection) connection).writeConnection.getUnderlyingConnection() + : connection; + return singleNodeConnection.createStatement(); + } + public static String defaultFormatDataTime(long time) { return RpcUtils.formatDatetime( RpcUtils.DEFAULT_TIME_FORMAT, TIME_PRECISION_IN_MS, time, DEFAULT_ZONE_ID); diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/enhanced/IoTDBPipeAutoDropIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/enhanced/IoTDBPipeAutoDropIT.java index eb060cf3f497..56623ef468a6 100644 --- a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/enhanced/IoTDBPipeAutoDropIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/enhanced/IoTDBPipeAutoDropIT.java @@ -22,10 +22,12 @@ import org.apache.iotdb.common.rpc.thrift.TSStatus; import org.apache.iotdb.commons.client.sync.SyncConfigNodeIServiceClient; import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStaticMeta; -import org.apache.iotdb.commons.schema.column.ColumnHeaderConstant; import org.apache.iotdb.confignode.rpc.thrift.TCreatePipeReq; +import org.apache.iotdb.confignode.rpc.thrift.TShowPipeInfo; +import org.apache.iotdb.confignode.rpc.thrift.TShowPipeReq; import org.apache.iotdb.consensus.ConsensusFactory; import org.apache.iotdb.db.it.utils.TestUtils; +import org.apache.iotdb.isession.SessionConfig; import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper; import org.apache.iotdb.it.framework.IoTDBTestRunner; import org.apache.iotdb.itbase.category.MultiClusterIT2DualTableManualEnhanced; @@ -41,16 +43,15 @@ import org.junit.runner.RunWith; import java.sql.Connection; -import java.sql.ResultSet; import java.sql.SQLException; import java.sql.Statement; import java.util.Collections; import java.util.HashMap; +import java.util.List; import java.util.Map; import java.util.concurrent.TimeUnit; import java.util.function.Consumer; -import static org.apache.iotdb.util.MagicUtils.makeItCloseQuietly; import static org.awaitility.Awaitility.await; import static org.junit.Assert.fail; @@ -125,10 +126,9 @@ public void testAutoDropInHistoricalTransfer() throws Exception { "test", handleFailure); - try (final Connection connection = - makeItCloseQuietly(senderEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT)); - final Statement statement = makeItCloseQuietly(connection.createStatement()); ) { - ResultSet result = statement.executeQuery("show pipes"); + try (final SyncConfigNodeIServiceClient client = + (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) { + // Fetch a fresh table-model snapshot on every poll while history pipes are being removed. await() .pollInSameThread() .pollDelay(1L, TimeUnit.SECONDS) @@ -136,19 +136,18 @@ public void testAutoDropInHistoricalTransfer() throws Exception { .atMost(600, TimeUnit.SECONDS) .untilAsserted( () -> { - try { - int pipeNum = 0; - while (result.next()) { - final String pipeName = result.getString(ColumnHeaderConstant.ID); - if (!pipeName.contains(PipeStaticMeta.CONSENSUS_PIPE_PREFIX) - && pipeName.endsWith("_history")) { - pipeNum++; - } - } - Assert.assertEquals(0, pipeNum); - } catch (Exception e) { - Assert.fail(); - } + final List showPipeResult = + client.showPipe( + new TShowPipeReq() + .setIsTableModel(true) + .setUserName(SessionConfig.DEFAULT_USER)) + .pipeInfoList; + Assert.assertTrue( + showPipeResult.stream() + .noneMatch( + i -> + !i.getId().startsWith(PipeStaticMeta.CONSENSUS_PIPE_PREFIX) + && i.getId().endsWith("_history"))); }); } } @@ -207,9 +206,8 @@ public void testAutoDropInHistoricalTransferWithTimeRange() throws Exception { handleFailure); } - try (final Connection connection = makeItCloseQuietly(senderEnv.getConnection()); - final Statement statement = makeItCloseQuietly(connection.createStatement()); ) { - ResultSet result = statement.executeQuery("show pipes"); + try (final SyncConfigNodeIServiceClient client = + (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) { await() .pollInSameThread() .pollDelay(1L, TimeUnit.SECONDS) @@ -217,19 +215,15 @@ public void testAutoDropInHistoricalTransferWithTimeRange() throws Exception { .atMost(600, TimeUnit.SECONDS) .untilAsserted( () -> { - try { - int pipeNum = 0; - while (result.next()) { - if (!result - .getString(ColumnHeaderConstant.ID) - .contains(PipeStaticMeta.CONSENSUS_PIPE_PREFIX)) { - pipeNum++; - } - } - Assert.assertEquals(0, pipeNum); - } catch (Exception e) { - Assert.fail(); - } + final List showPipeResult = + client.showPipe( + new TShowPipeReq() + .setIsTableModel(true) + .setUserName(SessionConfig.DEFAULT_USER)) + .pipeInfoList; + showPipeResult.removeIf( + i -> i.getId().startsWith(PipeStaticMeta.CONSENSUS_PIPE_PREFIX)); + Assert.assertTrue(showPipeResult.isEmpty()); }); } } diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/enhanced/IoTDBPipeAutoDropIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/enhanced/IoTDBPipeAutoDropIT.java index fa6694e3be7a..001cf24631c1 100644 --- a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/enhanced/IoTDBPipeAutoDropIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/enhanced/IoTDBPipeAutoDropIT.java @@ -22,7 +22,6 @@ import org.apache.iotdb.common.rpc.thrift.TSStatus; import org.apache.iotdb.commons.client.sync.SyncConfigNodeIServiceClient; import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStaticMeta; -import org.apache.iotdb.commons.schema.column.ColumnHeaderConstant; import org.apache.iotdb.confignode.rpc.thrift.TCreatePipeReq; import org.apache.iotdb.confignode.rpc.thrift.TShowPipeInfo; import org.apache.iotdb.confignode.rpc.thrift.TShowPipeReq; @@ -41,9 +40,6 @@ import org.junit.experimental.categories.Category; import org.junit.runner.RunWith; -import java.sql.Connection; -import java.sql.ResultSet; -import java.sql.Statement; import java.util.Arrays; import java.util.Collections; import java.util.HashMap; @@ -52,7 +48,6 @@ import java.util.Objects; import java.util.concurrent.TimeUnit; -import static org.apache.iotdb.util.MagicUtils.makeItCloseQuietly; import static org.awaitility.Awaitility.await; @RunWith(IoTDBTestRunner.class) @@ -115,31 +110,21 @@ public void testAutoDropInHistoricalTransfer() throws Exception { "count(root.db.d1.s1),", Collections.singleton("1,")); - try (final Connection connection = makeItCloseQuietly(senderEnv.getConnection()); - final Statement statement = makeItCloseQuietly(connection.createStatement()); ) { - ResultSet result = statement.executeQuery("show pipes"); - await() - .pollInSameThread() - .pollDelay(1L, TimeUnit.SECONDS) - .pollInterval(1L, TimeUnit.SECONDS) - .atMost(600, TimeUnit.SECONDS) - .untilAsserted( - () -> { - try { - int pipeNum = 0; - while (result.next()) { - if (!result - .getString(ColumnHeaderConstant.ID) - .contains(PipeStaticMeta.CONSENSUS_PIPE_PREFIX)) { - pipeNum++; - } - } - Assert.assertEquals(0, pipeNum); - } catch (Exception e) { - Assert.fail(); - } - }); - } + // Fetch a fresh ConfigNode snapshot on every poll while pipes are being removed. + await() + .pollInSameThread() + .pollDelay(1L, TimeUnit.SECONDS) + .pollInterval(1L, TimeUnit.SECONDS) + .atMost(600, TimeUnit.SECONDS) + .untilAsserted( + () -> { + final List showPipeResult = + client.showPipe(new TShowPipeReq().setUserName(SessionConfig.DEFAULT_USER)) + .pipeInfoList; + showPipeResult.removeIf( + i -> i.getId().startsWith(PipeStaticMeta.CONSENSUS_PIPE_PREFIX)); + Assert.assertTrue(showPipeResult.isEmpty()); + }); } } @@ -371,31 +356,20 @@ public void testAutoDropInHistoricalTransferWithTimeRange() throws Exception { "count(root.db.d1.s1),", Collections.singleton("3,")); - try (final Connection connection = makeItCloseQuietly(senderEnv.getConnection()); - final Statement statement = makeItCloseQuietly(connection.createStatement()); ) { - ResultSet result = statement.executeQuery("show pipes"); - await() - .pollInSameThread() - .pollDelay(1L, TimeUnit.SECONDS) - .pollInterval(1L, TimeUnit.SECONDS) - .atMost(600, TimeUnit.SECONDS) - .untilAsserted( - () -> { - try { - int pipeNum = 0; - while (result.next()) { - if (!result - .getString(ColumnHeaderConstant.ID) - .contains(PipeStaticMeta.CONSENSUS_PIPE_PREFIX)) { - pipeNum++; - } - } - Assert.assertEquals(0, pipeNum); - } catch (Exception e) { - Assert.fail(); - } - }); - } + await() + .pollInSameThread() + .pollDelay(1L, TimeUnit.SECONDS) + .pollInterval(1L, TimeUnit.SECONDS) + .atMost(600, TimeUnit.SECONDS) + .untilAsserted( + () -> { + final List showPipeResult = + client.showPipe(new TShowPipeReq().setUserName(SessionConfig.DEFAULT_USER)) + .pipeInfoList; + showPipeResult.removeIf( + i -> i.getId().startsWith(PipeStaticMeta.CONSENSUS_PIPE_PREFIX)); + Assert.assertTrue(showPipeResult.isEmpty()); + }); } } } 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 c6063a2c59d8..f82671f1ece4 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,7 +29,6 @@ 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; @@ -1357,7 +1356,8 @@ public void testDeleteTable() throws SQLException { int testNum = 12; prepareData(testNum, 1); try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT); - Statement statement = connection.createStatement()) { + Statement statement = connection.createStatement(); + Statement procedureStatement = TestUtils.createSingleNodeStatement(connection)) { statement.execute("use test"); statement.execute("DROP TABLE vehicle" + testNum); @@ -1366,12 +1366,10 @@ public void testDeleteTable() throws SQLException { 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( + procedureStatement.executeQuery( "SELECT state, error_message FROM information_schema.drop_table_procedures " + "WHERE database = 'test' AND table_name = 'vehicle" + testNum 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 a417373daa51..cdb7254b8c7a 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 @@ -803,11 +803,12 @@ public void testInformationSchema() throws SQLException { try (final Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT); - final Statement statement = connection.createStatement()) { + final Statement statement = connection.createStatement(); + final Statement diagnosticStatement = TestUtils.createSingleNodeStatement(connection)) { // Test table query statement.execute("use information_schema"); - try (final ResultSet resultSet = statement.executeQuery("SHOW PIPE MEMORY")) { + try (final ResultSet resultSet = diagnosticStatement.executeQuery("SHOW PIPE MEMORY")) { final ResultSetMetaData metaData = resultSet.getMetaData(); assertEquals(13, metaData.getColumnCount()); assertEquals("block_id", metaData.getColumnName(1)); @@ -835,7 +836,7 @@ public void testInformationSchema() throws SQLException { assertTrue(hasFloatingMemory); } try (final ResultSet resultSet = - statement.executeQuery("select * from information_schema.pipe_memory")) { + diagnosticStatement.executeQuery("select * from information_schema.pipe_memory")) { boolean hasFloatingMemory = false; while (resultSet.next()) { if ("FloatingMemory".equals(resultSet.getString(2))) { diff --git a/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBDropTableModProgressIT.java b/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBDropTableModProgressIT.java index fe00da05a05c..aa1b5ec27fd1 100644 --- a/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBDropTableModProgressIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBDropTableModProgressIT.java @@ -67,7 +67,8 @@ public static void tearDown() throws Exception { public void testDropTableModFileProgress() throws Exception { try (final Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT); - final Statement statement = connection.createStatement()) { + final Statement statement = connection.createStatement(); + final Statement procedureStatement = TestUtils.createSingleNodeStatement(connection)) { statement.execute("create database async_drop_mod_db"); statement.execute("use async_drop_mod_db"); statement.execute("create table drop_target (device string tag, reading int32)"); @@ -83,7 +84,7 @@ public void testDropTableModFileProgress() throws Exception { .untilAsserted( () -> { try (final ResultSet resultSet = - statement.executeQuery( + procedureStatement.executeQuery( "select state, written_mod_file_count, remaining_mod_file_count " + "from information_schema.drop_table_procedures " + "where database = 'async_drop_mod_db' and table_name = 'drop_target'")) { 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 ef59faff72c6..52146294997a 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 @@ -26,7 +26,6 @@ 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.runtime.ClusterTestConnection; import org.apache.iotdb.rpc.IoTDBConnectionException; import org.apache.iotdb.rpc.StatementExecutionException; @@ -85,7 +84,7 @@ public void testManageTable() { try (final Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT); final Statement statement = connection.createStatement(); - final Statement procedureStatement = createProcedureStatement(connection)) { + final Statement procedureStatement = TestUtils.createSingleNodeStatement(connection)) { statement.execute("create database test1"); statement.execute("create database test2 with (ttl=3000000)"); @@ -1265,7 +1264,7 @@ public void testAsyncDropTableProgress() throws Exception { try (final Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT); final Statement statement = connection.createStatement(); - final Statement procedureStatement = createProcedureStatement(connection)) { + final Statement procedureStatement = TestUtils.createSingleNodeStatement(connection)) { statement.execute("create database async_drop_db"); statement.execute("use async_drop_db"); statement.execute("create table drop_target (device string tag, reading int32)"); @@ -1298,17 +1297,6 @@ public void testAsyncDropTableProgress() throws Exception { } } - private static Statement createProcedureStatement(final Connection connection) - throws SQLException { - // Procedure progress can change between queries on different DataNodes. Query a single node - // without the cluster connection's cross-node result verification. - final Connection procedureConnection = - connection instanceof ClusterTestConnection - ? ((ClusterTestConnection) connection).writeConnection.getUnderlyingConnection() - : connection; - return procedureConnection.createStatement(); - } - private static void awaitDropProcedure( final Statement statement, final String database, final String tableName) throws SQLException { From 743d85b41b58013730db013eaf6128448a6bbf0b Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Fri, 16 Oct 2026 10:23:05 +0800 Subject: [PATCH 3/6] [Test] Retry database cleanup after asynchronous table drop --- .../apache/iotdb/db/it/utils/TestUtils.java | 24 +++++++++++++++++++ .../manual/enhanced/IoTDBPipeMetaIT.java | 7 ++++-- 2 files changed, 29 insertions(+), 2 deletions(-) diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/utils/TestUtils.java b/integration-test/src/test/java/org/apache/iotdb/db/it/utils/TestUtils.java index aabce4f92f89..5540b0a0031c 100644 --- a/integration-test/src/test/java/org/apache/iotdb/db/it/utils/TestUtils.java +++ b/integration-test/src/test/java/org/apache/iotdb/db/it/utils/TestUtils.java @@ -30,6 +30,7 @@ import org.apache.iotdb.rpc.IoTDBConnectionException; import org.apache.iotdb.rpc.RpcUtils; import org.apache.iotdb.rpc.StatementExecutionException; +import org.apache.iotdb.rpc.TSStatusCode; import org.apache.tsfile.read.common.RowRecord; import org.junit.Assert; @@ -985,6 +986,29 @@ public static void executeTableNonQuery(String sql, String userName, String pass } } + public static void dropDatabaseWithRetry(final Statement statement, final String database) + throws SQLException { + // Asynchronous DROP TABLE may still be running when a test starts database cleanup. + final long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(60); + while (true) { + try { + statement.execute("DROP DATABASE IF EXISTS " + database); + return; + } catch (final SQLException e) { + if (e.getErrorCode() != TSStatusCode.OVERLAP_WITH_EXISTING_TASK.getStatusCode() + || System.nanoTime() >= deadline) { + throw e; + } + } + try { + Thread.sleep(100); + } catch (final InterruptedException e) { + Thread.currentThread().interrupt(); + throw new SQLException(e); + } + } + } + public static void executeNonQueryWithRetry(final BaseEnv env, final String sql) { executeNonQueryWithRetry(env, sql, SessionConfig.DEFAULT_USER, SessionConfig.DEFAULT_PASSWORD); } diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/enhanced/IoTDBPipeMetaIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/enhanced/IoTDBPipeMetaIT.java index 86135d293dee..99d2be35188b 100644 --- a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/enhanced/IoTDBPipeMetaIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/enhanced/IoTDBPipeMetaIT.java @@ -184,8 +184,11 @@ public void testTableSync() throws Exception { Collections.emptySet(), dbName); - TestUtils.executeNonQuery( - dbName, BaseEnv.TABLE_SQL_DIALECT, senderEnv, "drop database test", null); + // Receiver metadata can disappear before the sender's asynchronous drop has finished. + try (Connection connection = senderEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT); + Statement statement = connection.createStatement()) { + TestUtils.dropDatabaseWithRetry(statement, dbName); + } // Will not include no-transfer database TestUtils.assertDataEventuallyOnEnv( From 877c016b4ed2ce167f4bca147d5a333f11c0f919 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Fri, 16 Oct 2026 11:45:01 +0800 Subject: [PATCH 4/6] [Test] Fix remaining drop table waits and continuous assertions --- .../iotdb/db/it/utils/IoTDBTestUtilsIT.java | 129 ++++++++++++++++++ .../apache/iotdb/db/it/utils/TestUtils.java | 82 ++++++++--- .../manual/basic/IoTDBPipePermissionIT.java | 8 +- .../manual/basic/IoTDBPipeWithLoadIT.java | 2 + .../manual/enhanced/IoTDBPipeMetaIT.java | 13 +- .../manual/IoTDBPipeInclusionIT.java | 26 ++-- .../manual/IoTDBPipeMetaHistoricalIT.java | 6 +- .../manual/IoTDBPipePermissionIT.java | 12 +- .../it/db/it/IoTDBDeletionTableIT.java | 21 +-- .../it/insertquery/IoTDBInsertQueryIT.java | 3 +- .../it/query/recent/IoTDBCteIT.java | 5 +- .../it/query/recent/IoTExplainAnalyzeIT.java | 5 +- .../it/schema/IoTDBTableDDLHAIT.java | 42 ++---- .../relational/it/schema/IoTDBTableIT.java | 40 +----- 14 files changed, 255 insertions(+), 139 deletions(-) create mode 100644 integration-test/src/test/java/org/apache/iotdb/db/it/utils/IoTDBTestUtilsIT.java diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/utils/IoTDBTestUtilsIT.java b/integration-test/src/test/java/org/apache/iotdb/db/it/utils/IoTDBTestUtilsIT.java new file mode 100644 index 000000000000..80d00c3843b3 --- /dev/null +++ b/integration-test/src/test/java/org/apache/iotdb/db/it/utils/IoTDBTestUtilsIT.java @@ -0,0 +1,129 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.db.it.utils; + +import org.apache.iotdb.it.env.EnvFactory; +import org.apache.iotdb.it.framework.IoTDBTestRunner; +import org.apache.iotdb.itbase.category.TableClusterIT; +import org.apache.iotdb.itbase.category.TableLocalStandaloneIT; +import org.apache.iotdb.itbase.env.BaseEnv; + +import org.junit.AfterClass; +import org.junit.BeforeClass; +import org.junit.Test; +import org.junit.experimental.categories.Category; +import org.junit.runner.RunWith; + +import java.util.Collections; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.Consumer; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertThrows; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; + +@RunWith(IoTDBTestRunner.class) +@Category({TableLocalStandaloneIT.class, TableClusterIT.class}) +public class IoTDBTestUtilsIT { + + @BeforeClass + public static void setUp() throws Exception { + EnvFactory.getEnv().initClusterEnvironment(); + } + + @AfterClass + public static void tearDown() { + EnvFactory.getEnv().cleanClusterEnvironment(); + } + + @Test + public void testAssertAlwaysFail() { + assertObservationPeriod( + () -> + TestUtils.assertAlwaysFail( + EnvFactory.getEnv(), "select * from missing_database.missing_table", 1)); + assertThrows( + AssertionError.class, + () -> TestUtils.assertAlwaysFail(EnvFactory.getEnv(), "show databases", 1)); + } + + @Test + public void testAssertDataAlwaysFailsOnMismatch() { + final BaseEnv env = EnvFactory.getEnv(); + final AtomicInteger failures = new AtomicInteger(); + final Consumer handleFailure = message -> failures.incrementAndGet(); + assertThrows( + AssertionError.class, + () -> + TestUtils.assertDataAlwaysOnEnv( + env, "count databases", "unexpected,", Collections.emptySet(), 1, (String) null)); + assertThrows( + AssertionError.class, + () -> + TestUtils.assertDataAlwaysOnEnv( + env, "count databases", "unexpected,", Collections.emptySet(), 1, handleFailure)); + assertThrows( + AssertionError.class, + () -> + TestUtils.assertDataAlwaysOnEnv( + env, + "show databases", + "unexpected,", + Collections.emptySet(), + 1, + "information_schema", + handleFailure)); + assertEquals(2, failures.get()); + } + + @Test + public void testAssertDataAlwaysObservationPeriod() { + final BaseEnv env = EnvFactory.getEnv(); + final Consumer handleFailure = message -> fail(message); + assertObservationPeriod( + () -> + TestUtils.assertDataAlwaysOnEnv( + env, "count databases", "count,", Collections.singleton("0,"), 1, (String) null)); + assertObservationPeriod( + () -> + TestUtils.assertDataAlwaysOnEnv( + env, "count databases", "count,", Collections.singleton("0,"), 1, handleFailure)); + assertObservationPeriod( + () -> + TestUtils.assertDataAlwaysOnEnv( + env, + "count databases", + "count,", + Collections.singleton("1,"), + 1, + "information_schema", + handleFailure)); + } + + private void assertObservationPeriod(final Runnable assertion) { + final long start = System.nanoTime(); + assertion.run(); + assertTrue( + "The assertion must observe queries for the requested period", + System.nanoTime() - start >= TimeUnit.SECONDS.toNanos(1)); + } +} diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/utils/TestUtils.java b/integration-test/src/test/java/org/apache/iotdb/db/it/utils/TestUtils.java index 5540b0a0031c..acf39b6ab6b9 100644 --- a/integration-test/src/test/java/org/apache/iotdb/db/it/utils/TestUtils.java +++ b/integration-test/src/test/java/org/apache/iotdb/db/it/utils/TestUtils.java @@ -100,6 +100,38 @@ public static Statement createSingleNodeStatement(final Connection connection) return singleNodeConnection.createStatement(); } + /** Waits for the latest DROP TABLE procedure using a single DataNode's progress snapshot. */ + public static void awaitDropTableProcedure( + final Connection connection, final String database, final String tableName) + throws SQLException { + try (final Statement statement = createSingleNodeStatement(connection)) { + await() + .atMost(60, TimeUnit.SECONDS) + .pollInSameThread() + .pollInterval(20, TimeUnit.MILLISECONDS) + .until( + () -> { + try (final ResultSet resultSet = + statement.executeQuery( + "SELECT state, error_message FROM information_schema.drop_table_procedures " + + "WHERE database = '" + + database.replace("'", "''") + + "' AND table_name = '" + + tableName.replace("'", "''") + + "' ORDER BY procedure_id DESC LIMIT 1")) { + if (!resultSet.next()) { + return false; + } + final String state = resultSet.getString(1); + if ("ROLLEDBACK".equals(state)) { + throw new SQLException(resultSet.getString(2)); + } + return "SUCCESS".equals(state); + } + }); + } + } + public static String defaultFormatDataTime(long time) { return RpcUtils.formatDatetime( RpcUtils.DEFAULT_TIME_FORMAT, TIME_PRECISION_IN_MS, time, DEFAULT_ZONE_ID); @@ -1896,7 +1928,7 @@ public static void assertDataEventuallyOnEnv( } } - // Note that this class will accept any exceptions + /** Asserts that a query keeps failing with an SQLException throughout the observation period. */ public static void assertAlwaysFail(final BaseEnv env, final String sql) { assertAlwaysFail(env, sql, 10); } @@ -1905,13 +1937,15 @@ public static void assertAlwaysFail( final BaseEnv env, final String sql, final long consistentSeconds) { try (final Connection connection = env.getConnection(BaseEnv.TABLE_SQL_DIALECT); final Statement statement = connection.createStatement()) { - // Keep retrying if there are execution failures await() .pollInSameThread() - .pollDelay(1L, TimeUnit.SECONDS) + .pollDelay(0L, TimeUnit.SECONDS) .pollInterval(1L, TimeUnit.SECONDS) - .atMost(consistentSeconds, TimeUnit.SECONDS) - .failFast(() -> Assert.assertThrows(Exception.class, () -> statement.executeQuery(sql))); + .during(consistentSeconds, TimeUnit.SECONDS) + .atMost(consistentSeconds + 5, TimeUnit.SECONDS) + .failFast( + () -> Assert.assertThrows(SQLException.class, () -> statement.executeQuery(sql))) + .until(() -> true); } catch (final Exception e) { fail(e.getMessage()); } @@ -2044,12 +2078,12 @@ public static void assertDataAlwaysOnEnv( env.getConnection( Objects.isNull(database) ? TREE_SQL_DIALECT : BaseEnv.TABLE_SQL_DIALECT); Statement statement = connection.createStatement()) { - // Keep retrying if there are execution failures await() .pollInSameThread() - .pollDelay(1L, TimeUnit.SECONDS) + .pollDelay(0L, TimeUnit.SECONDS) .pollInterval(1L, TimeUnit.SECONDS) - .atMost(consistentSeconds, TimeUnit.SECONDS) + .during(consistentSeconds, TimeUnit.SECONDS) + .atMost(consistentSeconds + 5, TimeUnit.SECONDS) .failFast( () -> { try { @@ -2060,7 +2094,8 @@ public static void assertDataAlwaysOnEnv( // sync requires tsFile to process. We flush in the middle of assertion because we // don't know when the data reaches the receiver in general cases if (!flushed[0] - && System.currentTimeMillis() - startTime > consistentSeconds >> 1) { + && System.currentTimeMillis() - startTime + > TimeUnit.SECONDS.toMillis(consistentSeconds) / 2) { flushed[0] = true; statement.execute("flush"); } @@ -2069,7 +2104,8 @@ public static void assertDataAlwaysOnEnv( } catch (Exception e) { Assert.fail(); } - }); + }) + .until(() -> true); } catch (Exception e) { e.printStackTrace(); fail(); @@ -2089,9 +2125,10 @@ public static void assertDataAlwaysOnEnv( Statement statement = connection.createStatement()) { await() .pollInSameThread() - .pollDelay(1L, TimeUnit.SECONDS) + .pollDelay(0L, TimeUnit.SECONDS) .pollInterval(1L, TimeUnit.SECONDS) - .atMost(consistentSeconds, TimeUnit.SECONDS) + .during(consistentSeconds, TimeUnit.SECONDS) + .atMost(consistentSeconds + 5, TimeUnit.SECONDS) .failFast( () -> { try { @@ -2099,7 +2136,8 @@ public static void assertDataAlwaysOnEnv( // sync requires tsFile to process. We flush in the middle of assertion because we // don't know when the data reaches the receiver in general cases if (!flushed[0] - && System.currentTimeMillis() - startTime > consistentSeconds >> 1) { + && System.currentTimeMillis() - startTime + > TimeUnit.SECONDS.toMillis(consistentSeconds) / 2) { flushed[0] = true; statement.execute("flush"); } @@ -2116,7 +2154,8 @@ public static void assertDataAlwaysOnEnv( } throw e; } - }); + }) + .until(() -> true); } catch (Exception e) { e.printStackTrace(); fail(); @@ -2133,14 +2172,15 @@ public static void assertDataAlwaysOnEnv( Consumer handleFailure) { final long startTime = System.currentTimeMillis(); final boolean[] flushed = {false}; - try (Connection connection = env.getConnection(); + try (Connection connection = + env.getConnection(database == null ? TREE_SQL_DIALECT : BaseEnv.TABLE_SQL_DIALECT); Statement statement = connection.createStatement()) { - // Keep retrying if there are execution failures await() .pollInSameThread() - .pollDelay(1L, TimeUnit.SECONDS) + .pollDelay(0L, TimeUnit.SECONDS) .pollInterval(1L, TimeUnit.SECONDS) - .atMost(consistentSeconds, TimeUnit.SECONDS) + .during(consistentSeconds, TimeUnit.SECONDS) + .atMost(consistentSeconds + 5, TimeUnit.SECONDS) .failFast( () -> { try { @@ -2151,7 +2191,8 @@ public static void assertDataAlwaysOnEnv( // sync requires tsFile to process. We flush in the middle of assertion because we // don't know when the data reaches the receiver in general cases if (!flushed[0] - && System.currentTimeMillis() - startTime > consistentSeconds >> 1) { + && System.currentTimeMillis() - startTime + > TimeUnit.SECONDS.toMillis(consistentSeconds) / 2) { flushed[0] = true; statement.execute("flush"); } @@ -2168,7 +2209,8 @@ public static void assertDataAlwaysOnEnv( } throw e; } - }); + }) + .until(() -> true); } catch (Exception e) { e.printStackTrace(); fail(); diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/basic/IoTDBPipePermissionIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/basic/IoTDBPipePermissionIT.java index c47777b189f9..0cda218010a6 100644 --- a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/basic/IoTDBPipePermissionIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/basic/IoTDBPipePermissionIT.java @@ -324,7 +324,7 @@ public void testReceiverPermission() throws Exception { "show databases", "Database,TTL(ms),SchemaReplicationFactor,DataReplicationFactor,TimePartitionInterval,", Collections.singleton("information_schema,INF,null,null,null,"), - (String) null); + "information_schema"); TestUtils.executeNonQuery( "information_schema", @@ -374,7 +374,11 @@ public void testReceiverPermission() throws Exception { // Shall not be transferred TestUtils.assertDataAlwaysOnEnv( - receiverEnv, "count databases", "count,", Collections.singleton("2,"), (String) null); + receiverEnv, + "count databases", + "count,", + Collections.singleton("2,"), + "information_schema"); TestUtils.executeNonQuery( "information_schema", diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/basic/IoTDBPipeWithLoadIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/basic/IoTDBPipeWithLoadIT.java index d635a9a41805..c3a6c259379e 100644 --- a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/basic/IoTDBPipeWithLoadIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/basic/IoTDBPipeWithLoadIT.java @@ -415,6 +415,8 @@ public void testLoadAutoCreateWithTableDeletion() throws Exception { statement.execute("INSERT INTO t1(time,tag1,tag2,s1,s2) values(2, 'd1', 'd2', 'blue', 2)"); statement.execute("flush"); statement.execute("drop table t1"); + // Start historical transfer only after the table and its data have been deleted. + TestUtils.awaitDropTableProcedure(connection, "db", "t1"); } catch (Exception e) { fail(e.getMessage()); } diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/enhanced/IoTDBPipeMetaIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/enhanced/IoTDBPipeMetaIT.java index 99d2be35188b..506f6facc944 100644 --- a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/enhanced/IoTDBPipeMetaIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/enhanced/IoTDBPipeMetaIT.java @@ -241,7 +241,7 @@ public void testNoTree() throws Exception { TestUtils.assertDataAlwaysOnEnv( receiverEnv, "show databases", - "Database,SchemaReplicationFactor,DataReplicationFactor,TimePartitionInterval,", + "Database,SchemaReplicationFactor,DataReplicationFactor,TimePartitionOrigin,TimePartitionInterval,", Collections.emptySet()); } } @@ -299,7 +299,7 @@ public void testNoTable() throws Exception { "show databases", "Database,TTL(ms),SchemaReplicationFactor,DataReplicationFactor,TimePartitionInterval,", Collections.singleton("information_schema,INF,null,null,null,"), - dbName); + "information_schema"); } } @@ -362,18 +362,17 @@ public void testAuth() throws Exception { "grant alter on any to user testUser with grant option", null); - TestUtils.assertDataAlwaysOnEnv( + TestUtils.assertDataEventuallyOnEnv( receiverEnv, "list privileges of user testUser", "Role,Scope,Privileges,GrantOption,", new HashSet<>( Arrays.asList( - ",,MANAGE_USER,false,", - ",,MANAGE_ROLE,false,", - ",,MAINTAIN,false,", + ",,SYSTEM,false,", + ",,SECURITY,false,", ",*.*,ALTER,true,", ",test.*,DROP,false,")), - (String) null); + "information_schema"); } } diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/manual/IoTDBPipeInclusionIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/manual/IoTDBPipeInclusionIT.java index a95e8535d422..0366687849ca 100644 --- a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/manual/IoTDBPipeInclusionIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/manual/IoTDBPipeInclusionIT.java @@ -108,7 +108,10 @@ public void testPureSchemaInclusion() throws Exception { "root.ln.wf01.wt01.status,null,root.ln,BOOLEAN,PLAIN,ZSTD,{\"tag3\":\"v3\"},{\"attr4\":\"v4\"},null,null,BASE,")); TestUtils.assertDataAlwaysOnEnv( - receiverEnv, "select * from root.ln.**", "Time,", Collections.emptySet()); + receiverEnv, + "select * from root.ln.**", + "Time,root.ln.wf01.wt01.status,", + Collections.emptySet()); } } @@ -179,8 +182,12 @@ public void testPureSchemaInclusionWithMultiplePattern() throws Exception { "insert into root.ln.wf01.wt01(time, status) values(now(), false)", "flush"), null); + // Keep the header independent of wildcard expansion order. TestUtils.assertDataAlwaysOnEnv( - receiverEnv, "select * from root.ln.**", "Time,", Collections.emptySet()); + receiverEnv, + "select wf01.wt01.status, wf02.wt01.status from root.ln", + "Time,root.ln.wf01.wt01.status,root.ln.wf02.wt01.status,", + Collections.emptySet()); } } @@ -254,7 +261,10 @@ public void testPureSchemaInclusionWithExclusionPattern() throws Exception { null); TestUtils.assertDataAlwaysOnEnv( - receiverEnv, "select * from root.ln.**", "Time,", Collections.emptySet()); + receiverEnv, + "select * from root.ln.**", + "Time,root.ln.wf01.wt01.status,", + Collections.emptySet()); } } @@ -293,7 +303,7 @@ public void testAuthExclusion() throws Exception { TestUtils.executeNonQuery(senderEnv, "create user `ln_write_user` 'write_pwd123456'", null); TestUtils.assertDataAlwaysOnEnv( - receiverEnv, "list user", "user,", Collections.singleton("root,")); + receiverEnv, "list user", "UserId,User,", Collections.singleton("0,root,")); } } @@ -337,14 +347,14 @@ public void testAuthInclusionWithPattern() throws Exception { "GRANT READ_DATA, WRITE_DATA ON root.** TO USER ln_write_user;"), null); - TestUtils.assertDataAlwaysOnEnv( + TestUtils.assertDataEventuallyOnEnv( receiverEnv, "LIST PRIVILEGES OF USER ln_write_user", - "ROLE,PATH,PRIVILEGES,GRANT OPTION,", + "Role,Scope,Privileges,GrantOption,", new HashSet<>( Arrays.asList( - ",root.**,SYSTEM,true,", - ",root.**,SECURITY,true,", + ",,SYSTEM,true,", + ",,SECURITY,true,", ",root.ln.**,READ_DATA,false,", ",root.ln.**,WRITE_DATA,false,"))); } diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/manual/IoTDBPipeMetaHistoricalIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/manual/IoTDBPipeMetaHistoricalIT.java index 06b4a72ccf7d..6df47f39f2ac 100644 --- a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/manual/IoTDBPipeMetaHistoricalIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/manual/IoTDBPipeMetaHistoricalIT.java @@ -147,8 +147,8 @@ public void testTemplateInclusion() throws Exception { TestUtils.assertDataAlwaysOnEnv( receiverEnv, "list user", - ColumnHeaderConstant.USER + ",", - Collections.singleton("root,")); + ColumnHeaderConstant.USER_ID + "," + ColumnHeaderConstant.USER + ",", + Collections.singleton("0,root,")); TestUtils.assertDataAlwaysOnEnv( receiverEnv, "list role", ColumnHeaderConstant.ROLE + ",", Collections.emptySet()); @@ -259,7 +259,7 @@ public void testAuthInclusion() throws Exception { "Database,SchemaReplicationFactor,DataReplicationFactor,TimePartitionOrigin,TimePartitionInterval,", Collections.emptySet()); TestUtils.assertDataAlwaysOnEnv( - receiverEnv, "select * from root.ln.**", "Time", Collections.emptySet()); + receiverEnv, "select * from root.ln.**", "Time,", Collections.emptySet()); TestUtils.executeNonQuery(senderEnv, "CREATE ROLE test", null); diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/manual/IoTDBPipePermissionIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/manual/IoTDBPipePermissionIT.java index f2d5faa6e93f..b9fc3906292f 100644 --- a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/manual/IoTDBPipePermissionIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/manual/IoTDBPipePermissionIT.java @@ -526,15 +526,15 @@ public void testSourcePermission() { TestUtils.assertDataAlwaysOnEnv( receiverEnv, "select count(temperature) from root.vehicle.car", - "count(root.vehicle.car.pressure),", + "count(root.vehicle.car.temperature),", Collections.singleton("0,")); - // Exception, skip + // Without the series schema, the query has only a Time column and no rows. TestUtils.assertDataAlwaysOnEnv( receiverEnv, "select count(temperature) from root.vehicle.plane", - "count(root.vehicle.plane.temperature),", - Collections.singleton("0,")); + "Time,", + Collections.emptySet()); // Alter pipe, throw exception if no privileges try (final Connection connection = senderEnv.getConnection(); @@ -556,8 +556,8 @@ public void testSourcePermission() { TestUtils.assertDataAlwaysOnEnv( receiverEnv, "select count(pressure) from root.vehicle.car", - "count(root.vehicle.car.pressure),", - Collections.singleton("0,")); + "Time,", + Collections.emptySet()); // Grant SELECT privilege TestUtils.executeNonQueries( 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 f82671f1ece4..1aa186ffa2a8 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 @@ -137,7 +137,7 @@ public void setUp() { public void tearDown() { try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT); Statement statement = connection.createStatement()) { - statement.execute("DROP DATABASE IF EXISTS test"); + TestUtils.dropDatabaseWithRetry(statement, "test"); } catch (Exception e) { fail(e.getMessage()); } @@ -1356,28 +1356,13 @@ public void testDeleteTable() throws SQLException { int testNum = 12; prepareData(testNum, 1); try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT); - Statement statement = connection.createStatement(); - Statement procedureStatement = TestUtils.createSingleNodeStatement(connection)) { + Statement statement = connection.createStatement()) { statement.execute("use test"); statement.execute("DROP TABLE vehicle" + testNum); // DROP TABLE acknowledges submission before the table has been fully removed. - Awaitility.await() - .atMost(60, TimeUnit.SECONDS) - .pollInSameThread() - .untilAsserted( - () -> { - try (ResultSet resultSet = - procedureStatement.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)); - } - }); + TestUtils.awaitDropTableProcedure(connection, "test", "vehicle" + testNum); try (ResultSet ignored = statement.executeQuery("SELECT * FROM vehicle" + testNum)) { fail("Exception expected"); diff --git a/integration-test/src/test/java/org/apache/iotdb/relational/it/insertquery/IoTDBInsertQueryIT.java b/integration-test/src/test/java/org/apache/iotdb/relational/it/insertquery/IoTDBInsertQueryIT.java index 174b2d4711ae..75a7725c4149 100644 --- a/integration-test/src/test/java/org/apache/iotdb/relational/it/insertquery/IoTDBInsertQueryIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/relational/it/insertquery/IoTDBInsertQueryIT.java @@ -19,6 +19,7 @@ package org.apache.iotdb.relational.it.insertquery; +import org.apache.iotdb.db.it.utils.TestUtils; import org.apache.iotdb.isession.ITableSession; import org.apache.iotdb.isession.SessionDataSet; import org.apache.iotdb.it.env.EnvFactory; @@ -99,7 +100,7 @@ public void setUp() throws SQLException { public void tearDown() { try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT); Statement statement = connection.createStatement()) { - statement.execute("DROP DATABASE IF EXISTS test"); + TestUtils.dropDatabaseWithRetry(statement, "test"); } catch (Exception e) { fail(e.getMessage()); } diff --git a/integration-test/src/test/java/org/apache/iotdb/relational/it/query/recent/IoTDBCteIT.java b/integration-test/src/test/java/org/apache/iotdb/relational/it/query/recent/IoTDBCteIT.java index 8c0b52acc944..65d3f277e52d 100644 --- a/integration-test/src/test/java/org/apache/iotdb/relational/it/query/recent/IoTDBCteIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/relational/it/query/recent/IoTDBCteIT.java @@ -19,6 +19,7 @@ package org.apache.iotdb.relational.it.query.recent; +import org.apache.iotdb.db.it.utils.TestUtils; import org.apache.iotdb.isession.ITableSession; import org.apache.iotdb.isession.SessionConfig; import org.apache.iotdb.isession.SessionDataSet; @@ -80,8 +81,6 @@ public class IoTDBCteIT { "INSERT INTO testtb VALUES(1000, 300.0, 'c', 'd2')", }; - private static final String dropDbSqls = "DROP DATABASE IF EXISTS testdb"; - private static final String[] cteKeywords = {"", "materialized"}; @BeforeClass @@ -110,7 +109,7 @@ public void setUp() throws SQLException { public void tearDown() { try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT); Statement statement = connection.createStatement()) { - statement.execute(dropDbSqls); + TestUtils.dropDatabaseWithRetry(statement, "testdb"); } catch (Exception e) { fail(e.getMessage()); } diff --git a/integration-test/src/test/java/org/apache/iotdb/relational/it/query/recent/IoTExplainAnalyzeIT.java b/integration-test/src/test/java/org/apache/iotdb/relational/it/query/recent/IoTExplainAnalyzeIT.java index 4c252dbf3ca1..a36f01338fe6 100644 --- a/integration-test/src/test/java/org/apache/iotdb/relational/it/query/recent/IoTExplainAnalyzeIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/relational/it/query/recent/IoTExplainAnalyzeIT.java @@ -21,6 +21,7 @@ package org.apache.iotdb.relational.it.query.recent; +import org.apache.iotdb.db.it.utils.TestUtils; import org.apache.iotdb.it.env.EnvFactory; import org.apache.iotdb.it.framework.IoTDBTestRunner; import org.apache.iotdb.itbase.category.TableLocalStandaloneIT; @@ -63,8 +64,6 @@ public class IoTExplainAnalyzeIT { "INSERT INTO testtb VALUES(1000, 'd2', 300.0)", }; - private static final String dropDbSqls = "DROP DATABASE IF EXISTS " + DATABASE_NAME; - @BeforeClass public static void setUpClass() { Locale.setDefault(Locale.ENGLISH); @@ -92,7 +91,7 @@ public void setUp() throws SQLException { public void tearDown() { try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT); Statement statement = connection.createStatement()) { - statement.execute(dropDbSqls); + TestUtils.dropDatabaseWithRetry(statement, DATABASE_NAME); } catch (Exception e) { fail(e.getMessage()); } diff --git a/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBTableDDLHAIT.java b/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBTableDDLHAIT.java index 59d90f6d9fae..ed0786c23eb1 100644 --- a/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBTableDDLHAIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBTableDDLHAIT.java @@ -21,15 +21,14 @@ import org.apache.iotdb.commons.cluster.NodeStatus; import org.apache.iotdb.consensus.ConsensusFactory; +import org.apache.iotdb.db.it.utils.TestUtils; import org.apache.iotdb.isession.SessionConfig; import org.apache.iotdb.it.env.EnvFactory; import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper; import org.apache.iotdb.it.framework.IoTDBTestRunner; import org.apache.iotdb.itbase.category.TableClusterIT; import org.apache.iotdb.itbase.env.BaseEnv; -import org.apache.iotdb.itbase.exception.InconsistentDataException; -import org.awaitility.Awaitility; import org.junit.Assert; import org.junit.Test; import org.junit.experimental.categories.Category; @@ -43,7 +42,6 @@ import java.sql.Statement; import java.util.Collections; import java.util.concurrent.Callable; -import java.util.concurrent.TimeUnit; import static org.junit.Assert.assertTrue; @@ -121,7 +119,8 @@ public void testHAWithOneDataNodeIsDown() throws Exception { victimDataNode.stop(); Assert.assertFalse("victim DataNode should be stopped", victimDataNode.isAlive()); - executeSpecificSql(databaseName, tableName, statement, createdAfterDownTableName); + executeSpecificSql( + databaseName, tableName, connection, statement, createdAfterDownTableName); testDropDeviceForTreeModel(statement); } } finally { @@ -169,7 +168,8 @@ public void testHAWithOneDataNodeIsReadOnly() throws Exception { // Run DDL HA tests via live DataNode. Operations should still succeed because // consensus can proceed with 2/3 nodes, and the ReadOnly victim still accepts // committed entries as a follower. - executeSpecificSql(databaseName, tableName, statement, createdAfterDownTableName); + executeSpecificSql( + databaseName, tableName, connection, statement, createdAfterDownTableName); } } finally { @@ -178,7 +178,11 @@ public void testHAWithOneDataNodeIsReadOnly() throws Exception { } public void executeSpecificSql( - String databaseName, String tableName, Statement statement, String createdAfterDownTableName) + String databaseName, + String tableName, + Connection connection, + Statement statement, + String createdAfterDownTableName) throws Exception { // Take one DataNode down. Its last successful ConfigNode contact is now frozen; after @@ -233,7 +237,7 @@ public void executeSpecificSql( LOGGER.info("6. start to test high availability of dropping table procedure"); statement.execute("DROP TABLE " + tableName); // DROP TABLE acknowledges submission before the table has been fully removed. - awaitDropTableProcedure(statement, databaseName, tableName); + TestUtils.awaitDropTableProcedure(connection, databaseName, tableName); assertTrue( "DROP TABLE must succeed with one DataNode down", !tableExists(statement, tableName)); @@ -277,30 +281,6 @@ private void assertStatementEffect( assertTrue(message, effect.call()); } - private void awaitDropTableProcedure( - final Statement statement, final String databaseName, final String tableName) - throws Exception { - Awaitility.await() - .atMost(60, TimeUnit.SECONDS) - .pollInSameThread() - // Progress snapshots can differ between cluster nodes while deletion is running. - .ignoreException(InconsistentDataException.class) - .untilAsserted( - () -> { - try (final ResultSet resultSet = - statement.executeQuery( - "SELECT state, error_message FROM information_schema.drop_table_procedures " - + "WHERE database = '" - + databaseName - + "' AND table_name = '" - + tableName - + "' ORDER BY procedure_id DESC LIMIT 1")) { - assertTrue("DROP TABLE procedure should be visible", resultSet.next()); - Assert.assertEquals(resultSet.getString(2), "SUCCESS", resultSet.getString(1)); - } - }); - } - private boolean tableExists(final Statement statement, final String tableName) throws Exception { try (final ResultSet resultSet = statement.executeQuery("SHOW TABLES")) { while (resultSet.next()) { 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 52146294997a..ce4327297efe 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 @@ -83,8 +83,7 @@ public static void tearDown() throws Exception { public void testManageTable() { try (final Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT); - final Statement statement = connection.createStatement(); - final Statement procedureStatement = TestUtils.createSingleNodeStatement(connection)) { + final Statement statement = connection.createStatement()) { statement.execute("create database test1"); statement.execute("create database test2 with (ttl=3000000)"); @@ -547,7 +546,7 @@ public void testManageTable() { Collections.singleton("null,")); statement.execute("drop table table2"); - awaitDropProcedure(procedureStatement, "test2", "table2"); + TestUtils.awaitDropTableProcedure(connection, "test2", "table2"); try { statement.executeQuery("describe table2"); fail(); @@ -1272,7 +1271,7 @@ public void testAsyncDropTableProgress() throws Exception { // 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(procedureStatement, "async_drop_db", "drop_target"); + TestUtils.awaitDropTableProcedure(connection, "async_drop_db", "drop_target"); TestUtils.assertResultSetEqual( statement.executeQuery("show tables from async_drop_db"), "TableName,TTL(ms),", @@ -1296,37 +1295,4 @@ public void testAsyncDropTableProgress() throws Exception { } } } - - 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"); - } } From f15d24542935966c450f99b67bcdcb5de1bcf3ff Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Fri, 16 Oct 2026 12:27:24 +0800 Subject: [PATCH 5/6] [Test] Synchronize table drops and fix review feedback --- .../apache/iotdb/db/it/utils/TestUtils.java | 5 ++-- .../manual/basic/IoTDBPipeWithLoadIT.java | 3 +-- .../manual/enhanced/IoTDBPipeAutoDropIT.java | 25 +++++++++++-------- .../auto/basic/IoTDBPipeLifeCycleIT.java | 6 ++--- .../auto/enhanced/IoTDBPipeAutoDropIT.java | 17 ++++++++----- .../it/db/it/IoTDBDeletionTableIT.java | 5 +--- .../it/schema/IoTDBTableDDLHAIT.java | 4 +-- .../relational/it/schema/IoTDBTableIT.java | 7 ++---- 8 files changed, 37 insertions(+), 35 deletions(-) diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/utils/TestUtils.java b/integration-test/src/test/java/org/apache/iotdb/db/it/utils/TestUtils.java index acf39b6ab6b9..e559fe9fb189 100644 --- a/integration-test/src/test/java/org/apache/iotdb/db/it/utils/TestUtils.java +++ b/integration-test/src/test/java/org/apache/iotdb/db/it/utils/TestUtils.java @@ -100,11 +100,12 @@ public static Statement createSingleNodeStatement(final Connection connection) return singleNodeConnection.createStatement(); } - /** Waits for the latest DROP TABLE procedure using a single DataNode's progress snapshot. */ - public static void awaitDropTableProcedure( + /** Drops a table and waits for the asynchronous procedure to finish on a single DataNode. */ + public static void dropTable( final Connection connection, final String database, final String tableName) throws SQLException { try (final Statement statement = createSingleNodeStatement(connection)) { + statement.execute("DROP TABLE " + database + "." + tableName); await() .atMost(60, TimeUnit.SECONDS) .pollInSameThread() diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/basic/IoTDBPipeWithLoadIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/basic/IoTDBPipeWithLoadIT.java index c3a6c259379e..fe90aedf647b 100644 --- a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/basic/IoTDBPipeWithLoadIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/basic/IoTDBPipeWithLoadIT.java @@ -414,9 +414,8 @@ public void testLoadAutoCreateWithTableDeletion() throws Exception { statement.execute("INSERT INTO t1(time,tag1,tag2,s1,s2) values(1, 'd1', 'd2', 'red', 1)"); statement.execute("INSERT INTO t1(time,tag1,tag2,s1,s2) values(2, 'd1', 'd2', 'blue', 2)"); statement.execute("flush"); - statement.execute("drop table t1"); // Start historical transfer only after the table and its data have been deleted. - TestUtils.awaitDropTableProcedure(connection, "db", "t1"); + TestUtils.dropTable(connection, "db", "t1"); } catch (Exception e) { fail(e.getMessage()); } diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/enhanced/IoTDBPipeAutoDropIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/enhanced/IoTDBPipeAutoDropIT.java index 56623ef468a6..335a3a7e8e0a 100644 --- a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/enhanced/IoTDBPipeAutoDropIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/enhanced/IoTDBPipeAutoDropIT.java @@ -25,6 +25,7 @@ import org.apache.iotdb.confignode.rpc.thrift.TCreatePipeReq; import org.apache.iotdb.confignode.rpc.thrift.TShowPipeInfo; import org.apache.iotdb.confignode.rpc.thrift.TShowPipeReq; +import org.apache.iotdb.confignode.rpc.thrift.TShowPipeResp; import org.apache.iotdb.consensus.ConsensusFactory; import org.apache.iotdb.db.it.utils.TestUtils; import org.apache.iotdb.isession.SessionConfig; @@ -136,12 +137,14 @@ public void testAutoDropInHistoricalTransfer() throws Exception { .atMost(600, TimeUnit.SECONDS) .untilAsserted( () -> { - final List showPipeResult = + final TShowPipeResp response = client.showPipe( - new TShowPipeReq() - .setIsTableModel(true) - .setUserName(SessionConfig.DEFAULT_USER)) - .pipeInfoList; + new TShowPipeReq() + .setIsTableModel(true) + .setUserName(SessionConfig.DEFAULT_USER)); + Assert.assertEquals( + TSStatusCode.SUCCESS_STATUS.getStatusCode(), response.getStatus().getCode()); + final List showPipeResult = response.getPipeInfoList(); Assert.assertTrue( showPipeResult.stream() .noneMatch( @@ -215,12 +218,14 @@ public void testAutoDropInHistoricalTransferWithTimeRange() throws Exception { .atMost(600, TimeUnit.SECONDS) .untilAsserted( () -> { - final List showPipeResult = + final TShowPipeResp response = client.showPipe( - new TShowPipeReq() - .setIsTableModel(true) - .setUserName(SessionConfig.DEFAULT_USER)) - .pipeInfoList; + new TShowPipeReq() + .setIsTableModel(true) + .setUserName(SessionConfig.DEFAULT_USER)); + Assert.assertEquals( + TSStatusCode.SUCCESS_STATUS.getStatusCode(), response.getStatus().getCode()); + final List showPipeResult = response.getPipeInfoList(); showPipeResult.removeIf( i -> i.getId().startsWith(PipeStaticMeta.CONSENSUS_PIPE_PREFIX)); Assert.assertTrue(showPipeResult.isEmpty()); diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeLifeCycleIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeLifeCycleIT.java index 188f2813115c..4373345900b4 100644 --- a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeLifeCycleIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeLifeCycleIT.java @@ -228,14 +228,14 @@ public void testLifeCycleWithHistoryDisabled() throws Exception { receiverEnv, "count timeseries root.ln0.**", "count(timeseries),", - Collections.singleton("1,")); + Collections.singleton("0,")); TestUtils.assertDataAlwaysOnEnv( receiverEnv, "count timeseries root.db.**", "count(timeseries),", - Collections.singleton("1,")); + Collections.singleton("2,")); TestUtils.assertDataAlwaysOnEnv( - receiverEnv, "count databases root.ln0", "count,", Collections.singleton("1,")); + receiverEnv, "count databases root.ln0", "count,", Collections.singleton("0,")); TestUtils.assertDataAlwaysOnEnv( receiverEnv, "count databases root.db", "count,", Collections.singleton("1,")); diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/enhanced/IoTDBPipeAutoDropIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/enhanced/IoTDBPipeAutoDropIT.java index 001cf24631c1..973e482f660e 100644 --- a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/enhanced/IoTDBPipeAutoDropIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/enhanced/IoTDBPipeAutoDropIT.java @@ -25,6 +25,7 @@ import org.apache.iotdb.confignode.rpc.thrift.TCreatePipeReq; import org.apache.iotdb.confignode.rpc.thrift.TShowPipeInfo; import org.apache.iotdb.confignode.rpc.thrift.TShowPipeReq; +import org.apache.iotdb.confignode.rpc.thrift.TShowPipeResp; import org.apache.iotdb.db.it.utils.TestUtils; import org.apache.iotdb.isession.SessionConfig; import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper; @@ -118,9 +119,11 @@ public void testAutoDropInHistoricalTransfer() throws Exception { .atMost(600, TimeUnit.SECONDS) .untilAsserted( () -> { - final List showPipeResult = - client.showPipe(new TShowPipeReq().setUserName(SessionConfig.DEFAULT_USER)) - .pipeInfoList; + final TShowPipeResp response = + client.showPipe(new TShowPipeReq().setUserName(SessionConfig.DEFAULT_USER)); + Assert.assertEquals( + TSStatusCode.SUCCESS_STATUS.getStatusCode(), response.getStatus().getCode()); + final List showPipeResult = response.getPipeInfoList(); showPipeResult.removeIf( i -> i.getId().startsWith(PipeStaticMeta.CONSENSUS_PIPE_PREFIX)); Assert.assertTrue(showPipeResult.isEmpty()); @@ -294,9 +297,11 @@ public void testAutoDropIgnoredUnmatchedDataRegions() throws Exception { .atMost(600, TimeUnit.SECONDS) .untilAsserted( () -> { - final List showPipeResult = - client.showPipe(new TShowPipeReq().setUserName(SessionConfig.DEFAULT_USER)) - .pipeInfoList; + final TShowPipeResp response = + client.showPipe(new TShowPipeReq().setUserName(SessionConfig.DEFAULT_USER)); + Assert.assertEquals( + TSStatusCode.SUCCESS_STATUS.getStatusCode(), response.getStatus().getCode()); + final List showPipeResult = response.getPipeInfoList(); showPipeResult.removeIf( i -> i.getId().startsWith(PipeStaticMeta.CONSENSUS_PIPE_PREFIX)); Assert.assertTrue( 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 1aa186ffa2a8..1ff1beebec24 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 @@ -1359,10 +1359,7 @@ public void testDeleteTable() throws SQLException { Statement statement = connection.createStatement()) { statement.execute("use test"); - statement.execute("DROP TABLE vehicle" + testNum); - - // DROP TABLE acknowledges submission before the table has been fully removed. - TestUtils.awaitDropTableProcedure(connection, "test", "vehicle" + testNum); + TestUtils.dropTable(connection, "test", "vehicle" + testNum); try (ResultSet ignored = statement.executeQuery("SELECT * FROM vehicle" + testNum)) { fail("Exception expected"); diff --git a/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBTableDDLHAIT.java b/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBTableDDLHAIT.java index ed0786c23eb1..39bcd6d46f54 100644 --- a/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBTableDDLHAIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBTableDDLHAIT.java @@ -235,9 +235,7 @@ public void executeSpecificSql( "DELETE DEVICES must succeed with one DataNode down"); LOGGER.info("6. start to test high availability of dropping table procedure"); - statement.execute("DROP TABLE " + tableName); - // DROP TABLE acknowledges submission before the table has been fully removed. - TestUtils.awaitDropTableProcedure(connection, databaseName, tableName); + TestUtils.dropTable(connection, databaseName, tableName); assertTrue( "DROP TABLE must succeed with one DataNode down", !tableExists(statement, tableName)); 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 ce4327297efe..25f4cdc66cf4 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 @@ -545,8 +545,7 @@ public void testManageTable() { "speed,", Collections.singleton("null,")); - statement.execute("drop table table2"); - TestUtils.awaitDropTableProcedure(connection, "test2", "table2"); + TestUtils.dropTable(connection, "test2", "table2"); try { statement.executeQuery("describe table2"); fail(); @@ -1267,11 +1266,9 @@ public void testAsyncDropTableProgress() throws Exception { 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. - TestUtils.awaitDropTableProcedure(connection, "async_drop_db", "drop_target"); + TestUtils.dropTable(connection, "async_drop_db", "drop_target"); TestUtils.assertResultSetEqual( statement.executeQuery("show tables from async_drop_db"), "TableName,TTL(ms),", From 3b565d1155646daad14bdc1ed87850b6264c61f2 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Fri, 16 Oct 2026 12:48:42 +0800 Subject: [PATCH 6/6] [Test] Check all tree pipe auto-drop polling response statuses --- .../auto/enhanced/IoTDBPipeAutoDropIT.java | 16 ++++++++++------ 1 file changed, 10 insertions(+), 6 deletions(-) diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/enhanced/IoTDBPipeAutoDropIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/enhanced/IoTDBPipeAutoDropIT.java index 973e482f660e..a3e5d32cb72a 100644 --- a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/enhanced/IoTDBPipeAutoDropIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/enhanced/IoTDBPipeAutoDropIT.java @@ -191,9 +191,11 @@ public void testAutoDropFinitePipesWithoutDataRegion() throws Exception { .atMost(600, TimeUnit.SECONDS) .untilAsserted( () -> { - final List pipeInfoList = - client.showPipe(new TShowPipeReq().setUserName(SessionConfig.DEFAULT_USER)) - .pipeInfoList; + final TShowPipeResp response = + client.showPipe(new TShowPipeReq().setUserName(SessionConfig.DEFAULT_USER)); + Assert.assertEquals( + TSStatusCode.SUCCESS_STATUS.getStatusCode(), response.getStatus().getCode()); + final List pipeInfoList = response.getPipeInfoList(); Assert.assertFalse( pipeInfoList.stream() .anyMatch(info -> info.getId().equals("query_pipe_without_data_region"))); @@ -368,9 +370,11 @@ public void testAutoDropInHistoricalTransferWithTimeRange() throws Exception { .atMost(600, TimeUnit.SECONDS) .untilAsserted( () -> { - final List showPipeResult = - client.showPipe(new TShowPipeReq().setUserName(SessionConfig.DEFAULT_USER)) - .pipeInfoList; + final TShowPipeResp response = + client.showPipe(new TShowPipeReq().setUserName(SessionConfig.DEFAULT_USER)); + Assert.assertEquals( + TSStatusCode.SUCCESS_STATUS.getStatusCode(), response.getStatus().getCode()); + final List showPipeResult = response.getPipeInfoList(); showPipeResult.removeIf( i -> i.getId().startsWith(PipeStaticMeta.CONSENSUS_PIPE_PREFIX)); Assert.assertTrue(showPipeResult.isEmpty());