This is an automated email from the ASF dual-hosted git repository.
JackieTien97 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new c9c16af42f6 [Test] Fix dynamic status queries and asynchronous
deletion polling (#18899)
c9c16af42f6 is described below
commit c9c16af42f696eb9fcecf287704c7825d2524e2c
Author: Caideyipi <[email protected]>
AuthorDate: Mon Oct 19 09:01:51 2026 +0800
[Test] Fix dynamic status queries and asynchronous deletion polling (#18899)
---
.../apache/iotdb/db/it/utils/IoTDBTestUtilsIT.java | 129 +++++++++++++++++++++
.../org/apache/iotdb/db/it/utils/TestUtils.java | 123 ++++++++++++++++----
.../manual/basic/IoTDBPipePermissionIT.java | 8 +-
.../manual/basic/IoTDBPipeWithLoadIT.java | 3 +-
.../manual/enhanced/IoTDBPipeAutoDropIT.java | 71 ++++++------
.../manual/enhanced/IoTDBPipeMetaIT.java | 20 ++--
.../treemodel/auto/basic/IoTDBPipeLifeCycleIT.java | 6 +-
.../auto/enhanced/IoTDBPipeAutoDropIT.java | 105 +++++++----------
.../treemodel/manual/IoTDBPipeInclusionIT.java | 26 +++--
.../manual/IoTDBPipeMetaHistoricalIT.java | 6 +-
.../treemodel/manual/IoTDBPipePermissionIT.java | 12 +-
.../relational/it/db/it/IoTDBDeletionTableIT.java | 24 +---
.../it/insertquery/IoTDBInsertQueryIT.java | 3 +-
.../relational/it/query/recent/IoTDBCteIT.java | 5 +-
.../it/query/recent/IoTExplainAnalyzeIT.java | 5 +-
.../relational/it/schema/IoTDBDatabaseIT.java | 7 +-
.../it/schema/IoTDBDropTableModProgressIT.java | 5 +-
.../relational/it/schema/IoTDBTableDDLHAIT.java | 44 ++-----
.../iotdb/relational/it/schema/IoTDBTableIT.java | 45 +------
19 files changed, 391 insertions(+), 256 deletions(-)
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 00000000000..80d00c3843b
--- /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<String> 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<String> 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 bf1cb4376bb..e559fe9fb18 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.EnvFactory;
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.
+ *
+ * <p>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 class TestUtils {
}
}
+ 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 class TestUtils {
}
}
- // 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 class TestUtils {
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 class TestUtils {
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 class TestUtils {
// 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 class TestUtils {
} catch (Exception e) {
Assert.fail();
}
- });
+ })
+ .until(() -> true);
} catch (Exception e) {
e.printStackTrace();
fail();
@@ -2049,9 +2126,10 @@ public class TestUtils {
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 class TestUtils {
// 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 class TestUtils {
}
throw e;
}
- });
+ })
+ .until(() -> true);
} catch (Exception e) {
e.printStackTrace();
fail();
@@ -2093,14 +2173,15 @@ public class TestUtils {
Consumer<String> 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 class TestUtils {
// 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 class TestUtils {
}
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 c47777b189f..0cda218010a 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 class IoTDBPipePermissionIT extends
AbstractPipeTableModelDualManualIT {
"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 class IoTDBPipePermissionIT extends
AbstractPipeTableModelDualManualIT {
// 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 d635a9a4180..fe90aedf647 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 class IoTDBPipeWithLoadIT extends
AbstractPipeTableModelDualManualIT {
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 eb060cf3f49..335a3a7e8e0 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 @@ package
org.apache.iotdb.pipe.it.dual.tablemodel.manual.enhanced;
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.experimental.categories.Category;
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 class IoTDBPipeAutoDropIT extends
AbstractPipeTableModelDualManualIT {
"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 class IoTDBPipeAutoDropIT extends
AbstractPipeTableModelDualManualIT {
.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<TShowPipeInfo> 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 class IoTDBPipeAutoDropIT extends
AbstractPipeTableModelDualManualIT {
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 class IoTDBPipeAutoDropIT extends
AbstractPipeTableModelDualManualIT {
.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<TShowPipeInfo> 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 86135d293de..506f6facc94 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 class IoTDBPipeMetaIT extends
AbstractPipeTableModelDualManualIT {
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 class IoTDBPipeMetaIT extends
AbstractPipeTableModelDualManualIT {
TestUtils.assertDataAlwaysOnEnv(
receiverEnv,
"show databases",
-
"Database,SchemaReplicationFactor,DataReplicationFactor,TimePartitionInterval,",
+
"Database,SchemaReplicationFactor,DataReplicationFactor,TimePartitionOrigin,TimePartitionInterval,",
Collections.emptySet());
}
}
@@ -296,7 +299,7 @@ public class IoTDBPipeMetaIT extends
AbstractPipeTableModelDualManualIT {
"show databases",
"Database,TTL(ms),SchemaReplicationFactor,DataReplicationFactor,TimePartitionInterval,",
Collections.singleton("information_schema,INF,null,null,null,"),
- dbName);
+ "information_schema");
}
}
@@ -359,18 +362,17 @@ public class IoTDBPipeMetaIT extends
AbstractPipeTableModelDualManualIT {
"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 188f2813115..4373345900b 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 class IoTDBPipeLifeCycleIT extends
AbstractPipeDualTreeModelAutoIT {
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 fa6694e3be7..a3e5d32cb72 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 @@ package
org.apache.iotdb.pipe.it.dual.treemodel.auto.enhanced;
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.Test;
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.Map;
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 class IoTDBPipeAutoDropIT extends
AbstractPipeDualTreeModelAutoIT {
"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<TShowPipeInfo> showPipeResult =
response.getPipeInfoList();
+ showPipeResult.removeIf(
+ i ->
i.getId().startsWith(PipeStaticMeta.CONSENSUS_PIPE_PREFIX));
+ Assert.assertTrue(showPipeResult.isEmpty());
+ });
}
}
@@ -203,9 +191,11 @@ public class IoTDBPipeAutoDropIT extends
AbstractPipeDualTreeModelAutoIT {
.atMost(600, TimeUnit.SECONDS)
.untilAsserted(
() -> {
- final List<TShowPipeInfo> 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<TShowPipeInfo> pipeInfoList =
response.getPipeInfoList();
Assert.assertFalse(
pipeInfoList.stream()
.anyMatch(info ->
info.getId().equals("query_pipe_without_data_region")));
@@ -309,9 +299,11 @@ public class IoTDBPipeAutoDropIT extends
AbstractPipeDualTreeModelAutoIT {
.atMost(600, TimeUnit.SECONDS)
.untilAsserted(
() -> {
- final List<TShowPipeInfo> 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<TShowPipeInfo> showPipeResult =
response.getPipeInfoList();
showPipeResult.removeIf(
i ->
i.getId().startsWith(PipeStaticMeta.CONSENSUS_PIPE_PREFIX));
Assert.assertTrue(
@@ -371,31 +363,22 @@ public class IoTDBPipeAutoDropIT extends
AbstractPipeDualTreeModelAutoIT {
"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<TShowPipeInfo> 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 a95e8535d42..0366687849c 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 class IoTDBPipeInclusionIT extends
AbstractPipeDualTreeModelManualIT {
"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 class IoTDBPipeInclusionIT extends
AbstractPipeDualTreeModelManualIT {
"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 class IoTDBPipeInclusionIT extends
AbstractPipeDualTreeModelManualIT {
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 class IoTDBPipeInclusionIT extends
AbstractPipeDualTreeModelManualIT {
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 class IoTDBPipeInclusionIT extends
AbstractPipeDualTreeModelManualIT {
"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 06b4a72ccf7..6df47f39f2a 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 class IoTDBPipeMetaHistoricalIT extends
AbstractPipeDualTreeModelManualIT
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 class IoTDBPipeMetaHistoricalIT extends
AbstractPipeDualTreeModelManualIT
"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 f2d5faa6e93..b9fc3906292 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 class IoTDBPipePermissionIT extends
AbstractPipeDualTreeModelManualIT {
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 class IoTDBPipePermissionIT extends
AbstractPipeDualTreeModelManualIT {
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 c6063a2c59d..1ff1beebec2 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.ManualIT;
import org.apache.iotdb.itbase.category.TableClusterIT;
import org.apache.iotdb.itbase.category.TableLocalStandaloneIT;
import org.apache.iotdb.itbase.env.BaseEnv;
-import org.apache.iotdb.itbase.exception.InconsistentDataException;
import org.apache.iotdb.itbase.exception.ParallelRequestTimeoutException;
import org.apache.iotdb.rpc.IoTDBConnectionException;
import org.apache.iotdb.rpc.StatementExecutionException;
@@ -138,7 +137,7 @@ public class IoTDBDeletionTableIT {
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 class IoTDBDeletionTableIT {
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 174b2d4711a..75a7725c414 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 class IoTDBInsertQueryIT {
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 8c0b52acc94..65d3f277e52 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 class IoTDBCteIT {
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 4c252dbf3ca..a36f01338fe 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 class IoTExplainAnalyzeIT {
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 a417373daa5..cdb7254b8c7 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 class IoTDBDatabaseIT {
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 class IoTDBDatabaseIT {
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 fe00da05a05..aa1b5ec27fd 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 class IoTDBDropTableModProgressIT {
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 class IoTDBDropTableModProgressIT {
.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 59d90f6d9fa..39bcd6d46f5 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 @@ package org.apache.iotdb.relational.it.schema;
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.SQLException;
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 class IoTDBTableDDLHAIT {
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 class IoTDBTableDDLHAIT {
// 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 class IoTDBTableDDLHAIT {
}
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 class IoTDBTableDDLHAIT {
"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 @@ public class IoTDBTableDDLHAIT {
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 34bd61d03d5..25f4cdc66cf 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 class IoTDBTableIT {
"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 class IoTDBTableIT {
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 class IoTDBTableIT {
}
}
}
-
- 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");
- }
}
