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 0000000000000..80d00c3843b3d --- /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 bf1cb4376bb95..e559fe9fb1897 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,9 +26,11 @@ 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; +import org.apache.iotdb.rpc.TSStatusCode; import org.apache.tsfile.read.common.RowRecord; import org.junit.Assert; @@ -83,6 +85,54 @@ 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(); + } + + /** 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() + .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); @@ -969,6 +1019,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); } @@ -1856,7 +1929,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); } @@ -1865,13 +1938,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()); } @@ -2004,12 +2079,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 { @@ -2020,7 +2095,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"); } @@ -2029,7 +2105,8 @@ public static void assertDataAlwaysOnEnv( } catch (Exception e) { Assert.fail(); } - }); + }) + .until(() -> true); } catch (Exception e) { e.printStackTrace(); fail(); @@ -2049,9 +2126,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 { @@ -2059,7 +2137,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"); } @@ -2076,7 +2155,8 @@ public static void assertDataAlwaysOnEnv( } throw e; } - }); + }) + .until(() -> true); } catch (Exception e) { e.printStackTrace(); fail(); @@ -2093,14 +2173,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 { @@ -2111,7 +2192,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"); } @@ -2128,7 +2210,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 c47777b189f97..0cda218010a64 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 d635a9a41805d..fe90aedf647b7 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,7 +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.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 eb060cf3f4976..335a3a7e8e0a5 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,13 @@ 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.confignode.rpc.thrift.TShowPipeResp; 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 +44,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 +127,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 +137,20 @@ 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 TShowPipeResp response = + client.showPipe( + 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( + i -> + !i.getId().startsWith(PipeStaticMeta.CONSENSUS_PIPE_PREFIX) + && i.getId().endsWith("_history"))); }); } } @@ -207,9 +209,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 +218,17 @@ 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 TShowPipeResp response = + client.showPipe( + 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/tablemodel/manual/enhanced/IoTDBPipeMetaIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/enhanced/IoTDBPipeMetaIT.java index 86135d293dee0..506f6facc944e 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( @@ -238,7 +241,7 @@ public void testNoTree() throws Exception { TestUtils.assertDataAlwaysOnEnv( receiverEnv, "show databases", - "Database,SchemaReplicationFactor,DataReplicationFactor,TimePartitionInterval,", + "Database,SchemaReplicationFactor,DataReplicationFactor,TimePartitionOrigin,TimePartitionInterval,", Collections.emptySet()); } } @@ -296,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"); } } @@ -359,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/auto/basic/IoTDBPipeLifeCycleIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeLifeCycleIT.java index 188f2813115c0..4373345900b48 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 fa6694e3be7a8..a3e5d32cb72a5 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,10 +22,10 @@ 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.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; @@ -41,9 +41,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 +49,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 +111,23 @@ 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 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()); + }); } } @@ -203,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"))); @@ -309,9 +299,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( @@ -371,31 +363,22 @@ 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 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()); + }); } } } 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 a95e8535d4223..0366687849ca1 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 06b4a72ccf7d2..6df47f39f2ac4 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 f2d5faa6e93f1..b9fc3906292f1 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 c6063a2c59d84..1ff1beebec241 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; @@ -138,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()); } @@ -1360,26 +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. - 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)); - } - }); + 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/insertquery/IoTDBInsertQueryIT.java b/integration-test/src/test/java/org/apache/iotdb/relational/it/insertquery/IoTDBInsertQueryIT.java index 174b2d4711ae8..75a7725c41498 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 8c0b52acc944b..65d3f277e52de 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 4c252dbf3ca1a..a36f01338fe6f 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/IoTDBDatabaseIT.java b/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBDatabaseIT.java index a417373daa515..cdb7254b8c7a5 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 fe00da05a05c1..aa1b5ec27fd1d 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/IoTDBTableDDLHAIT.java b/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBTableDDLHAIT.java index 59d90f6d9fae4..39bcd6d46f540 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 @@ -231,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. - awaitDropTableProcedure(statement, databaseName, tableName); + TestUtils.dropTable(connection, databaseName, tableName); assertTrue( "DROP TABLE must succeed with one DataNode down", !tableExists(statement, tableName)); @@ -277,30 +279,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 34bd61d03d523..25f4cdc66cf42 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"); - awaitDropProcedure(statement, "test2", "table2"); + TestUtils.dropTable(connection, "test2", "table2"); try { statement.executeQuery("describe table2"); fail(); @@ -1262,21 +1261,20 @@ 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 = 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)"); - 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.dropTable(connection, "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 " @@ -1294,37 +1292,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"); - } }