This is an automated email from the ASF dual-hosted git repository.
ericpai 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 9da5ad5281 [IOTDB-3463] IT framework supports read and compare from
multi data nodes (#6271)
9da5ad5281 is described below
commit 9da5ad5281e569646f6f32ec53ac6e399f37a8df
Author: BaiJian <[email protected]>
AuthorDate: Tue Jun 14 14:09:29 2022 +0800
[IOTDB-3463] IT framework supports read and compare from multi data nodes
(#6271)
---
.github/workflows/cluster-it.yml | 14 +-
integration-test/README.md | 27 +-
.../org/apache/iotdb/it/env/ClusterEnvBase.java | 111 +-
.../java/org/apache/iotdb/it/env/ConfigNode.java | 15 +-
.../java/org/apache/iotdb/it/env/DataNode.java | 13 +-
.../java/org/apache/iotdb/it/env/EnvFactory.java | 2 +
.../itbase/runtime/ClusterResultSetMetaData.java | 239 ++++
.../itbase/runtime/ClusterTestConnection.java | 333 ++++++
.../iotdb/itbase/runtime/ClusterTestResultSet.java | 1230 ++++++++++++++++++++
.../iotdb/itbase/runtime/ClusterTestStatement.java | 297 +++++
.../itbase/runtime/InconsistentDataException.java | 29 +
.../iotdb/itbase/runtime/NodeConnection.java | 85 ++
.../itbase/runtime/ParallelRequestDelegate.java | 60 +
.../iotdb/itbase/runtime/RequestDelegate.java | 86 ++
.../iotdb/itbase/runtime/RequestThreadPool.java | 49 +
.../itbase/runtime/SerialRequestDelegate.java | 47 +
metrics/pom.xml | 41 +
17 files changed, 2607 insertions(+), 71 deletions(-)
diff --git a/.github/workflows/cluster-it.yml b/.github/workflows/cluster-it.yml
index fa4a1f1fd9..724d818cfd 100644
--- a/.github/workflows/cluster-it.yml
+++ b/.github/workflows/cluster-it.yml
@@ -49,7 +49,19 @@ jobs:
shell: bash
# we do not compile client-cpp for saving time, it is tested in
client.yml
# we can skip influxdb-protocol because it has been tested separately
in influxdb-protocol.yml
- run: mvn -B clean verify -Dsession.test.skip=true
-Diotdb.test.skip=true -Dcluster.test.skip=true -Dtsfile.test.skip=true
-Dcommons.test.skip=true -Dconfignode.test.skip=true -Dconsensus.test.skip=true
-pl integration-test -am -PClusterIT
+ run: |
+ mvn clean verify \
+ -Dsession.test.skip=true \
+ -Diotdb.test.skip=true \
+ -Dcluster.test.skip=true \
+ -Dtsfile.test.skip=true \
+ -Dcommons.test.skip=true \
+ -Dconfignode.test.skip=true \
+ -Dconsensus.test.skip=true \
+ -Djdbc.test.skip=true \
+ -Dmetrics.test.skip=true \
+ -pl integration-test \
+ -am -PClusterIT
- name: Upload Artifact
if: failure()
uses: actions/upload-artifact@v3
diff --git a/integration-test/README.md b/integration-test/README.md
index 0aa5815a91..caa25e43c1 100644
--- a/integration-test/README.md
+++ b/integration-test/README.md
@@ -34,14 +34,25 @@ You can run the integration test in cluster mode. At
present, we have implemente
The maven command is:
```
-mvn clean verify -Dsession.test.skip=true -Diotdb.test.skip=true
-Dcluster.test.skip=true -Dtsfile.test.skip=true -Dcommons.test.skip=true
-Dconfignode.test.skip=true -Dconsensus.test.skip=true -pl integration-test -am
-PClusterIT
+mvn clean verify \
+-Dsession.test.skip=true \
+-Diotdb.test.skip=true \
+-Dcluster.test.skip=true \
+-Dtsfile.test.skip=true \
+-Dcommons.test.skip=true \
+-Dconfignode.test.skip=true \
+-Dconsensus.test.skip=true \
+-Djdbc.test.skip=true \
+-Dmetrics.test.skip=true \
+-pl integration-test \
+-am -PClusterIT
```
Notice that, this above maven command only run IT.
-------
Run in IntelliJ in cluster mode is so easy,
-- Step 0. Optionally, when you run the test for the first time, or when you
change the code of the module that the integration test module depends on, you
may need to use the following command to generate
`integration/target/template-node` for nodes of the pseudo cluster.
+- Step 0. Optionally, when you run the test for the first time, or when you
change the code of the module that the integration test module depends on, you
may need to use the following command to generate
`integration-test/target/template-node` for nodes of the pseudo cluster.
```
mvn clean package -pl integration-test -am -DskipTests -PClusterIT
```
@@ -64,7 +75,17 @@ Integration testing with local standalone mode can be run
with both maven and ID
The maven command is:
```
-mvn clean verify -Dsession.test.skip=true -Diotdb.test.skip=true
-Dcluster.test.skip=true -Dtsfile.test.skip=true -Dcommons.test.skip=true
-Dconfignode.test.skip=true -Dconsensus.test.skip=true -pl integration-test -am
+mvn clean verify \
+-Dsession.test.skip=true \
+-Diotdb.test.skip=true \
+-Dcluster.test.skip=true \
+-Dtsfile.test.skip=true \
+-Dcommons.test.skip=true \
+-Dconfignode.test.skip=true \
+-Dconsensus.test.skip=true \
+-Djdbc.test.skip=true \
+-Dmetrics.test.skip=true \
+ -pl integration-test -am
```
-------
diff --git
a/integration-test/src/main/java/org/apache/iotdb/it/env/ClusterEnvBase.java
b/integration-test/src/main/java/org/apache/iotdb/it/env/ClusterEnvBase.java
index 1fd072d760..a4b1bb2b1d 100644
--- a/integration-test/src/main/java/org/apache/iotdb/it/env/ClusterEnvBase.java
+++ b/integration-test/src/main/java/org/apache/iotdb/it/env/ClusterEnvBase.java
@@ -19,6 +19,8 @@
package org.apache.iotdb.it.env;
import org.apache.iotdb.itbase.env.BaseEnv;
+import org.apache.iotdb.itbase.runtime.ClusterTestConnection;
+import org.apache.iotdb.itbase.runtime.NodeConnection;
import org.apache.iotdb.jdbc.Config;
import org.apache.iotdb.jdbc.Constant;
import org.apache.iotdb.jdbc.IoTDBConnection;
@@ -35,6 +37,7 @@ import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
+import java.util.Random;
import static org.apache.iotdb.jdbc.Config.VERSION;
import static org.junit.Assert.fail;
@@ -43,6 +46,7 @@ public abstract class ClusterEnvBase implements BaseEnv {
private static final Logger logger =
LoggerFactory.getLogger(ClusterEnvBase.class);
private List<ConfigNode> configNodes;
private List<DataNode> dataNodes;
+ private final Random rand = new Random();
protected void initEnvironment(int configNodesNum, int dataNodesNum) throws
InterruptedException {
this.configNodes = new ArrayList<>();
@@ -133,14 +137,13 @@ public abstract class ClusterEnvBase implements BaseEnv {
statement.execute("SET STORAGE GROUP TO root.test" + counter);
statement.execute(
"CREATE TIMESERIES root.test" + counter + ".d0.s0 WITH
DATATYPE=INT32, ENCODING=RLE");
- if (statement.execute("SHOW TIMESERIES")) {
- ResultSet resultSet = statement.getResultSet();
+ try (ResultSet resultSet = statement.executeQuery("SHOW TIMESERIES")) {
if (resultSet.next()) {
statement.execute("DELETE STORAGE GROUP root.*");
+ logger.info("The whole cluster is ready.");
break;
}
}
-
} catch (SQLException e) {
logger.debug(counter + " time(s) connect to cluster failed!", e);
}
@@ -159,67 +162,67 @@ public abstract class ClusterEnvBase implements BaseEnv {
@Override
public Connection getConnection() throws SQLException {
- Connection connection = null;
-
- try {
- Class.forName(Config.JDBC_DRIVER_NAME);
- connection =
- DriverManager.getConnection(
- Config.IOTDB_URL_PREFIX
- + this.dataNodes.get(0).getIp()
- + ":"
- + this.dataNodes.get(0).getPort(),
- System.getProperty("User", "root"),
- System.getProperty("Password", "root"));
- } catch (ClassNotFoundException e) {
- e.printStackTrace();
- fail();
- }
- return connection;
+ return new ClusterTestConnection(getWriteConnection(null),
getReadConnections(null));
}
- public IoTDBConnection getConnection(int queryTimeout) throws SQLException {
- IoTDBConnection connection = null;
- try {
- Class.forName(Config.JDBC_DRIVER_NAME);
- connection =
- (IoTDBConnection)
- DriverManager.getConnection(
- Config.IOTDB_URL_PREFIX
- + this.dataNodes.get(0).getIp()
- + ":"
- + this.dataNodes.get(0).getPort(),
- System.getProperty("User", "root"),
- System.getProperty("Password", "root"));
- connection.setQueryTimeout(queryTimeout);
- } catch (ClassNotFoundException e) {
- e.printStackTrace();
- fail();
- }
+ private IoTDBConnection getConnection(int queryTimeout) throws SQLException {
+ IoTDBConnection connection =
+ (IoTDBConnection)
+ DriverManager.getConnection(
+ Config.IOTDB_URL_PREFIX
+ + this.dataNodes.get(0).getIp()
+ + ":"
+ + this.dataNodes.get(0).getPort(),
+ System.getProperty("User", "root"),
+ System.getProperty("Password", "root"));
+ connection.setQueryTimeout(queryTimeout);
return connection;
}
@Override
public Connection getConnection(Constant.Version version) throws
SQLException {
- Connection connection = null;
- try {
- Class.forName(Config.JDBC_DRIVER_NAME);
- connection =
+ return new ClusterTestConnection(getWriteConnection(version),
getReadConnections(version));
+ }
+
+ protected NodeConnection getWriteConnection(Constant.Version version) throws
SQLException {
+ // Randomly choose a node for handling write requests
+ DataNode dataNode =
this.dataNodes.get(rand.nextInt(this.dataNodes.size()));
+ String endpoint = dataNode.getIp() + ":" + dataNode.getPort();
+ Connection writeConnection =
+ DriverManager.getConnection(
+ Config.IOTDB_URL_PREFIX + endpoint + getVersionParam(version),
+ System.getProperty("User", "root"),
+ System.getProperty("Password", "root"));
+ return new NodeConnection(
+ endpoint,
+ NodeConnection.NodeRole.DATA_NODE,
+ NodeConnection.ConnectionRole.WRITE,
+ writeConnection);
+ }
+
+ protected List<NodeConnection> getReadConnections(Constant.Version version)
throws SQLException {
+ List<NodeConnection> readConnections = new ArrayList<>();
+ for (DataNode dataNode : this.dataNodes) {
+ String endpoint = dataNode.getIp() + ":" + dataNode.getPort();
+ Connection readConnection =
DriverManager.getConnection(
- Config.IOTDB_URL_PREFIX
- + this.dataNodes.get(0).getIp()
- + ":"
- + this.dataNodes.get(0).getPort()
- + "?"
- + VERSION
- + "="
- + version.toString(),
+ Config.IOTDB_URL_PREFIX + endpoint + getVersionParam(version),
System.getProperty("User", "root"),
System.getProperty("Password", "root"));
- } catch (ClassNotFoundException e) {
- e.printStackTrace();
- fail();
+ readConnections.add(
+ new NodeConnection(
+ endpoint,
+ NodeConnection.NodeRole.DATA_NODE,
+ NodeConnection.ConnectionRole.READ,
+ readConnection));
}
- return connection;
+ return readConnections;
+ }
+
+ private String getVersionParam(Constant.Version version) {
+ if (version == null) {
+ return "";
+ }
+ return "?" + VERSION + "=" + version;
}
}
diff --git
a/integration-test/src/main/java/org/apache/iotdb/it/env/ConfigNode.java
b/integration-test/src/main/java/org/apache/iotdb/it/env/ConfigNode.java
index 03781a6a28..1cfee88c2c 100644
--- a/integration-test/src/main/java/org/apache/iotdb/it/env/ConfigNode.java
+++ b/integration-test/src/main/java/org/apache/iotdb/it/env/ConfigNode.java
@@ -19,9 +19,10 @@
package org.apache.iotdb.it.env;
import java.io.File;
-import java.io.FileInputStream;
import java.io.FileWriter;
import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Paths;
import java.util.Properties;
import static org.junit.Assert.fail;
@@ -30,9 +31,9 @@ public class ConfigNode extends ClusterNodeBase {
private final int consensusPort;
private final String configPath;
- private final String targetConfignode;
+ private final String targetConfigNode;
- public ConfigNode(boolean isSeed, String targetConfignode, String testName) {
+ public ConfigNode(boolean isSeed, String targetConfigNode, String testName) {
int[] portList = super.searchAvailablePorts();
super.setPort(portList[0]);
@@ -83,9 +84,9 @@ public class ConfigNode extends ClusterNodeBase {
+ "iotdb-confignode.properties";
if (isSeed) {
- this.targetConfignode = getIpAndPortString();
+ this.targetConfigNode = getIpAndPortString();
} else {
- this.targetConfignode = targetConfignode;
+ this.targetConfigNode = targetConfigNode;
}
}
@@ -93,11 +94,11 @@ public class ConfigNode extends ClusterNodeBase {
public void changeConfig(Properties properties) {
try {
Properties configProperties = new Properties();
- configProperties.load(new FileInputStream(this.configPath));
+ configProperties.load(Files.newInputStream(Paths.get(this.configPath)));
configProperties.setProperty("rpc_address", super.getIp());
configProperties.setProperty("rpc_port",
String.valueOf(super.getPort()));
configProperties.setProperty("consensus_port",
String.valueOf(this.consensusPort));
- configProperties.setProperty("target_confignode", this.targetConfignode);
+ configProperties.setProperty("target_confignode", this.targetConfigNode);
configProperties.setProperty(
"schema_region_consensus_protocol_class",
"org.apache.iotdb.consensus.ratis.RatisConsensus");
diff --git
a/integration-test/src/main/java/org/apache/iotdb/it/env/DataNode.java
b/integration-test/src/main/java/org/apache/iotdb/it/env/DataNode.java
index e6d7a169ff..be56476396 100644
--- a/integration-test/src/main/java/org/apache/iotdb/it/env/DataNode.java
+++ b/integration-test/src/main/java/org/apache/iotdb/it/env/DataNode.java
@@ -19,16 +19,17 @@
package org.apache.iotdb.it.env;
import java.io.File;
-import java.io.FileInputStream;
import java.io.FileWriter;
import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Paths;
import java.util.Properties;
import static org.junit.Assert.fail;
public class DataNode extends ClusterNodeBase {
- private final String targetConfignode;
+ private final String targetConfigNode;
private final String configPath;
private final int dataBlockManagerPort;
@@ -36,9 +37,9 @@ public class DataNode extends ClusterNodeBase {
private final int dataRegionConsensusPort;
private final int schemaRegionConsensusPort;
- public DataNode(String targetConfignode, String testName) {
+ public DataNode(String targetConfigNode, String testName) {
- this.targetConfignode = targetConfignode;
+ this.targetConfigNode = targetConfigNode;
int[] portList = super.searchAvailablePorts();
super.setPort(portList[0]);
@@ -96,7 +97,7 @@ public class DataNode extends ClusterNodeBase {
public void changeConfig(Properties properties) {
try {
Properties configProperties = new Properties();
- configProperties.load(new FileInputStream(this.configPath));
+ configProperties.load(Files.newInputStream(Paths.get(this.configPath)));
configProperties.setProperty("rpc_address", super.getIp());
configProperties.setProperty("internal_ip", "127.0.0.1");
configProperties.setProperty("rpc_port",
String.valueOf(super.getPort()));
@@ -107,7 +108,7 @@ public class DataNode extends ClusterNodeBase {
"data_region_consensus_port",
String.valueOf(this.dataRegionConsensusPort));
configProperties.setProperty(
"schema_region_consensus_port",
String.valueOf(this.schemaRegionConsensusPort));
- configProperties.setProperty("config_nodes", this.targetConfignode);
+ configProperties.setProperty("config_nodes", this.targetConfigNode);
if (properties != null && !properties.isEmpty()) {
configProperties.putAll(properties);
}
diff --git
a/integration-test/src/main/java/org/apache/iotdb/it/env/EnvFactory.java
b/integration-test/src/main/java/org/apache/iotdb/it/env/EnvFactory.java
index 18ca4525a2..d9719c4a7f 100644
--- a/integration-test/src/main/java/org/apache/iotdb/it/env/EnvFactory.java
+++ b/integration-test/src/main/java/org/apache/iotdb/it/env/EnvFactory.java
@@ -19,6 +19,7 @@
package org.apache.iotdb.it.env;
import org.apache.iotdb.itbase.env.BaseEnv;
+import org.apache.iotdb.jdbc.Config;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -30,6 +31,7 @@ public class EnvFactory {
public static BaseEnv getEnv() {
if (env == null) {
try {
+ Class.forName(Config.JDBC_DRIVER_NAME);
logger.debug(">>>>>>>" + System.getProperty("TestEnv"));
switch (System.getProperty("TestEnv", "Standalone")) {
case "Standalone":
diff --git
a/integration-test/src/main/java/org/apache/iotdb/itbase/runtime/ClusterResultSetMetaData.java
b/integration-test/src/main/java/org/apache/iotdb/itbase/runtime/ClusterResultSetMetaData.java
new file mode 100644
index 0000000000..b66b817a32
--- /dev/null
+++
b/integration-test/src/main/java/org/apache/iotdb/itbase/runtime/ClusterResultSetMetaData.java
@@ -0,0 +1,239 @@
+/*
+ * 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.itbase.runtime;
+
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+import java.util.List;
+
+/** The implementation of {@link ResultSetMetaData} in cluster test. */
+public class ClusterResultSetMetaData implements ResultSetMetaData {
+
+ private final List<ResultSetMetaData> metaDataList;
+ private final List<String> endpoints;
+
+ public ClusterResultSetMetaData(List<ResultSetMetaData> metadataList,
List<String> endpoints) {
+ this.metaDataList = metadataList;
+ this.endpoints = endpoints;
+ }
+
+ @Override
+ public int getColumnCount() throws SQLException {
+ RequestDelegate<Integer> delegate = createRequestDelegate();
+ for (ResultSetMetaData rs : metaDataList) {
+ delegate.addRequest(rs::getColumnCount);
+ }
+ return delegate.requestAllAndCompare();
+ }
+
+ @Override
+ public boolean isAutoIncrement(int column) throws SQLException {
+ RequestDelegate<Boolean> delegate = createRequestDelegate();
+ for (ResultSetMetaData rs : metaDataList) {
+ delegate.addRequest(() -> rs.isAutoIncrement(column));
+ }
+ return delegate.requestAllAndCompare();
+ }
+
+ @Override
+ public boolean isCaseSensitive(int column) throws SQLException {
+ RequestDelegate<Boolean> delegate = createRequestDelegate();
+ for (ResultSetMetaData rs : metaDataList) {
+ delegate.addRequest(() -> rs.isCaseSensitive(column));
+ }
+ return delegate.requestAllAndCompare();
+ }
+
+ @Override
+ public boolean isSearchable(int column) throws SQLException {
+ RequestDelegate<Boolean> delegate = createRequestDelegate();
+ for (ResultSetMetaData rs : metaDataList) {
+ delegate.addRequest(() -> rs.isSearchable(column));
+ }
+ return delegate.requestAllAndCompare();
+ }
+
+ @Override
+ public boolean isCurrency(int column) throws SQLException {
+ RequestDelegate<Boolean> delegate = createRequestDelegate();
+ for (ResultSetMetaData rs : metaDataList) {
+ delegate.addRequest(() -> rs.isCurrency(column));
+ }
+ return delegate.requestAllAndCompare();
+ }
+
+ @Override
+ public int isNullable(int column) throws SQLException {
+ RequestDelegate<Integer> delegate = createRequestDelegate();
+ for (ResultSetMetaData rs : metaDataList) {
+ delegate.addRequest(() -> rs.isNullable(column));
+ }
+ return delegate.requestAllAndCompare();
+ }
+
+ @Override
+ public boolean isSigned(int column) throws SQLException {
+ RequestDelegate<Boolean> delegate = createRequestDelegate();
+ for (ResultSetMetaData rs : metaDataList) {
+ delegate.addRequest(() -> rs.isSigned(column));
+ }
+ return delegate.requestAllAndCompare();
+ }
+
+ @Override
+ public int getColumnDisplaySize(int column) throws SQLException {
+ RequestDelegate<Integer> delegate = createRequestDelegate();
+ for (ResultSetMetaData rs : metaDataList) {
+ delegate.addRequest(() -> rs.getColumnDisplaySize(column));
+ }
+ return delegate.requestAllAndCompare();
+ }
+
+ @Override
+ public String getColumnLabel(int column) throws SQLException {
+ RequestDelegate<String> delegate = createRequestDelegate();
+ for (ResultSetMetaData rs : metaDataList) {
+ delegate.addRequest(() -> rs.getColumnLabel(column));
+ }
+ return delegate.requestAllAndCompare();
+ }
+
+ @Override
+ public String getColumnName(int column) throws SQLException {
+ RequestDelegate<String> delegate = createRequestDelegate();
+ for (ResultSetMetaData rs : metaDataList) {
+ delegate.addRequest(() -> rs.getColumnName(column));
+ }
+ return delegate.requestAllAndCompare();
+ }
+
+ @Override
+ public String getSchemaName(int column) throws SQLException {
+ RequestDelegate<String> delegate = createRequestDelegate();
+ for (ResultSetMetaData rs : metaDataList) {
+ delegate.addRequest(() -> rs.getSchemaName(column));
+ }
+ return delegate.requestAllAndCompare();
+ }
+
+ @Override
+ public int getPrecision(int column) throws SQLException {
+ RequestDelegate<Integer> delegate = createRequestDelegate();
+ for (ResultSetMetaData rs : metaDataList) {
+ delegate.addRequest(() -> rs.getPrecision(column));
+ }
+ return delegate.requestAllAndCompare();
+ }
+
+ @Override
+ public int getScale(int column) throws SQLException {
+ RequestDelegate<Integer> delegate = createRequestDelegate();
+ for (ResultSetMetaData rs : metaDataList) {
+ delegate.addRequest(() -> rs.getScale(column));
+ }
+ return delegate.requestAllAndCompare();
+ }
+
+ @Override
+ public String getTableName(int column) throws SQLException {
+ RequestDelegate<String> delegate = createRequestDelegate();
+ for (ResultSetMetaData rs : metaDataList) {
+ delegate.addRequest(() -> rs.getTableName(column));
+ }
+ return delegate.requestAllAndCompare();
+ }
+
+ @Override
+ public String getCatalogName(int column) throws SQLException {
+ RequestDelegate<String> delegate = createRequestDelegate();
+ for (ResultSetMetaData rs : metaDataList) {
+ delegate.addRequest(() -> rs.getCatalogName(column));
+ }
+ return delegate.requestAllAndCompare();
+ }
+
+ @Override
+ public int getColumnType(int column) throws SQLException {
+ RequestDelegate<Integer> delegate = createRequestDelegate();
+ for (ResultSetMetaData rs : metaDataList) {
+ delegate.addRequest(() -> rs.getColumnType(column));
+ }
+ return delegate.requestAllAndCompare();
+ }
+
+ @Override
+ public String getColumnTypeName(int column) throws SQLException {
+ RequestDelegate<String> delegate = createRequestDelegate();
+ for (ResultSetMetaData rs : metaDataList) {
+ delegate.addRequest(() -> rs.getColumnTypeName(column));
+ }
+ return delegate.requestAllAndCompare();
+ }
+
+ @Override
+ public boolean isReadOnly(int column) throws SQLException {
+ RequestDelegate<Boolean> delegate = createRequestDelegate();
+ for (ResultSetMetaData rs : metaDataList) {
+ delegate.addRequest(() -> rs.isReadOnly(column));
+ }
+ return delegate.requestAllAndCompare();
+ }
+
+ @Override
+ public boolean isWritable(int column) throws SQLException {
+ RequestDelegate<Boolean> delegate = createRequestDelegate();
+ for (ResultSetMetaData rs : metaDataList) {
+ delegate.addRequest(() -> rs.isWritable(column));
+ }
+ return delegate.requestAllAndCompare();
+ }
+
+ @Override
+ public boolean isDefinitelyWritable(int column) throws SQLException {
+ RequestDelegate<Boolean> delegate = createRequestDelegate();
+ for (ResultSetMetaData rs : metaDataList) {
+ delegate.addRequest(() -> rs.isDefinitelyWritable(column));
+ }
+ return delegate.requestAllAndCompare();
+ }
+
+ @Override
+ public String getColumnClassName(int column) throws SQLException {
+ RequestDelegate<String> delegate = createRequestDelegate();
+ for (ResultSetMetaData rs : metaDataList) {
+ delegate.addRequest(() -> rs.getColumnClassName(column));
+ }
+ return delegate.requestAllAndCompare();
+ }
+
+ @Override
+ public <T> T unwrap(Class<T> iface) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public boolean isWrapperFor(Class<?> iface) {
+ throw new UnsupportedOperationException();
+ }
+
+ /** As all the ResultSetMetaData are on local, there's no need to request in
parallel */
+ private <T> RequestDelegate<T> createRequestDelegate() {
+ return new SerialRequestDelegate<>(endpoints);
+ }
+}
diff --git
a/integration-test/src/main/java/org/apache/iotdb/itbase/runtime/ClusterTestConnection.java
b/integration-test/src/main/java/org/apache/iotdb/itbase/runtime/ClusterTestConnection.java
new file mode 100644
index 0000000000..372c04a1b7
--- /dev/null
+++
b/integration-test/src/main/java/org/apache/iotdb/itbase/runtime/ClusterTestConnection.java
@@ -0,0 +1,333 @@
+/*
+ * 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.itbase.runtime;
+
+import org.apache.commons.lang3.Validate;
+
+import java.sql.Array;
+import java.sql.Blob;
+import java.sql.CallableStatement;
+import java.sql.Clob;
+import java.sql.Connection;
+import java.sql.DatabaseMetaData;
+import java.sql.NClob;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.sql.SQLWarning;
+import java.sql.SQLXML;
+import java.sql.Savepoint;
+import java.sql.Statement;
+import java.sql.Struct;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.Executor;
+
+/** The implementation of {@link Connection} in cluster test. */
+public class ClusterTestConnection implements Connection {
+
+ private final NodeConnection writeConnection;
+ private final List<NodeConnection> readConnections;
+ private boolean isClosed;
+
+ public ClusterTestConnection(
+ NodeConnection writeConnection, List<NodeConnection> readConnections) {
+ Validate.notNull(readConnections);
+ this.writeConnection = writeConnection;
+ this.readConnections = readConnections;
+ }
+
+ @Override
+ public Statement createStatement() throws SQLException {
+ return new ClusterTestStatement(writeConnection, readConnections);
+ }
+
+ @Override
+ public PreparedStatement prepareStatement(String sql) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public CallableStatement prepareCall(String sql) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public String nativeSQL(String sql) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void setAutoCommit(boolean autoCommit) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public boolean getAutoCommit() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void commit() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void rollback() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void close() {
+ writeConnection.close();
+ for (NodeConnection conn : readConnections) {
+ conn.close();
+ }
+ isClosed = true;
+ }
+
+ @Override
+ public boolean isClosed() {
+ return isClosed;
+ }
+
+ @Override
+ public DatabaseMetaData getMetaData() throws SQLException {
+ return writeConnection.getUnderlyingConnecton().getMetaData();
+ }
+
+ @Override
+ public void setReadOnly(boolean readOnly) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public boolean isReadOnly() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void setCatalog(String catalog) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public String getCatalog() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void setTransactionIsolation(int level) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public int getTransactionIsolation() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public SQLWarning getWarnings() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void clearWarnings() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Statement createStatement(int resultSetType, int
resultSetConcurrency) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public PreparedStatement prepareStatement(
+ String sql, int resultSetType, int resultSetConcurrency) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public CallableStatement prepareCall(String sql, int resultSetType, int
resultSetConcurrency) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Map<String, Class<?>> getTypeMap() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void setTypeMap(Map<String, Class<?>> map) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void setHoldability(int holdability) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public int getHoldability() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Savepoint setSavepoint() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Savepoint setSavepoint(String name) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void rollback(Savepoint savepoint) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void releaseSavepoint(Savepoint savepoint) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Statement createStatement(
+ int resultSetType, int resultSetConcurrency, int resultSetHoldability) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public PreparedStatement prepareStatement(
+ String sql, int resultSetType, int resultSetConcurrency, int
resultSetHoldability) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public CallableStatement prepareCall(
+ String sql, int resultSetType, int resultSetConcurrency, int
resultSetHoldability) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public PreparedStatement prepareStatement(String sql, int autoGeneratedKeys)
{
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public PreparedStatement prepareStatement(String sql, int[] columnIndexes) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public PreparedStatement prepareStatement(String sql, String[] columnNames) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Clob createClob() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Blob createBlob() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public NClob createNClob() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public SQLXML createSQLXML() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public boolean isValid(int timeout) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void setClientInfo(String name, String value) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void setClientInfo(Properties properties) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public String getClientInfo(String name) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Properties getClientInfo() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Array createArrayOf(String typeName, Object[] elements) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Struct createStruct(String typeName, Object[] attributes) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void setSchema(String schema) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public String getSchema() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void abort(Executor executor) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void setNetworkTimeout(Executor executor, int milliseconds) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public int getNetworkTimeout() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public <T> T unwrap(Class<T> iface) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public boolean isWrapperFor(Class<?> iface) {
+ throw new UnsupportedOperationException();
+ }
+}
diff --git
a/integration-test/src/main/java/org/apache/iotdb/itbase/runtime/ClusterTestResultSet.java
b/integration-test/src/main/java/org/apache/iotdb/itbase/runtime/ClusterTestResultSet.java
new file mode 100644
index 0000000000..a3fa82ca11
--- /dev/null
+++
b/integration-test/src/main/java/org/apache/iotdb/itbase/runtime/ClusterTestResultSet.java
@@ -0,0 +1,1230 @@
+/*
+ * 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.itbase.runtime;
+
+import java.io.InputStream;
+import java.io.Reader;
+import java.math.BigDecimal;
+import java.net.URL;
+import java.sql.Array;
+import java.sql.Blob;
+import java.sql.Clob;
+import java.sql.Date;
+import java.sql.NClob;
+import java.sql.Ref;
+import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
+import java.sql.RowId;
+import java.sql.SQLException;
+import java.sql.SQLWarning;
+import java.sql.SQLXML;
+import java.sql.Statement;
+import java.sql.Time;
+import java.sql.Timestamp;
+import java.util.Calendar;
+import java.util.List;
+import java.util.Map;
+
+/** The implementation of {@link ClusterTestResultSet} in cluster test. */
+public class ClusterTestResultSet implements ResultSet {
+
+ private final List<ResultSet> resultSets;
+ private final List<String> endpoints;
+ private final int queryTimeoutSeconds;
+
+ public ClusterTestResultSet(
+ List<Statement> statements, List<String> endpoints, String sql, int
queryTimeoutSeconds)
+ throws SQLException {
+ this.queryTimeoutSeconds = queryTimeoutSeconds;
+ this.endpoints = endpoints;
+ RequestDelegate<ResultSet> delegate = createRequestDelegate();
+ for (Statement st : statements) {
+ delegate.addRequest(() -> st.executeQuery(sql));
+ }
+ resultSets = delegate.requestAll();
+ }
+
+ @Override
+ public boolean next() throws SQLException {
+ RequestDelegate<Boolean> delegate = createRequestDelegate();
+ for (ResultSet rs : resultSets) {
+ delegate.addRequest(rs::next);
+ }
+ return delegate.requestAllAndCompare();
+ }
+
+ @Override
+ public void close() throws SQLException {
+ RequestDelegate<Void> delegate = createRequestDelegate();
+ for (ResultSet rs : resultSets) {
+ delegate.addRequest(
+ () -> {
+ rs.close();
+ return null;
+ });
+ }
+ delegate.requestAll();
+ }
+
+ @Override
+ public boolean wasNull() throws SQLException {
+ RequestDelegate<Boolean> delegate = createRequestDelegate();
+ for (ResultSet rs : resultSets) {
+ delegate.addRequest(rs::wasNull);
+ }
+ return delegate.requestAllAndCompare();
+ }
+
+ @Override
+ public String getString(int columnIndex) throws SQLException {
+ RequestDelegate<String> delegate = createRequestDelegate();
+ for (ResultSet rs : resultSets) {
+ delegate.addRequest(() -> rs.getString(columnIndex));
+ }
+ return delegate.requestAllAndCompare();
+ }
+
+ @Override
+ public boolean getBoolean(int columnIndex) throws SQLException {
+ RequestDelegate<Boolean> delegate = createRequestDelegate();
+ for (ResultSet rs : resultSets) {
+ delegate.addRequest(() -> rs.getBoolean(columnIndex));
+ }
+ return delegate.requestAllAndCompare();
+ }
+
+ @Override
+ public byte getByte(int columnIndex) throws SQLException {
+ RequestDelegate<Byte> delegate = createRequestDelegate();
+ for (ResultSet rs : resultSets) {
+ delegate.addRequest(() -> rs.getByte(columnIndex));
+ }
+ return delegate.requestAllAndCompare();
+ }
+
+ @Override
+ public short getShort(int columnIndex) throws SQLException {
+ RequestDelegate<Short> delegate = createRequestDelegate();
+ for (ResultSet rs : resultSets) {
+ delegate.addRequest(() -> rs.getShort(columnIndex));
+ }
+ return delegate.requestAllAndCompare();
+ }
+
+ @Override
+ public int getInt(int columnIndex) throws SQLException {
+ RequestDelegate<Integer> delegate = createRequestDelegate();
+ for (ResultSet rs : resultSets) {
+ delegate.addRequest(() -> rs.getInt(columnIndex));
+ }
+ return delegate.requestAllAndCompare();
+ }
+
+ @Override
+ public long getLong(int columnIndex) throws SQLException {
+ RequestDelegate<Long> delegate = createRequestDelegate();
+ for (ResultSet rs : resultSets) {
+ delegate.addRequest(() -> rs.getLong(columnIndex));
+ }
+ return delegate.requestAllAndCompare();
+ }
+
+ @Override
+ public float getFloat(int columnIndex) throws SQLException {
+ RequestDelegate<Float> delegate = createRequestDelegate();
+ for (ResultSet rs : resultSets) {
+ delegate.addRequest(() -> rs.getFloat(columnIndex));
+ }
+ return delegate.requestAllAndCompare();
+ }
+
+ @Override
+ public double getDouble(int columnIndex) throws SQLException {
+ RequestDelegate<Double> delegate = createRequestDelegate();
+ for (ResultSet rs : resultSets) {
+ delegate.addRequest(() -> rs.getDouble(columnIndex));
+ }
+ return delegate.requestAllAndCompare();
+ }
+
+ @Override
+ public BigDecimal getBigDecimal(int columnIndex, int scale) throws
SQLException {
+ RequestDelegate<BigDecimal> delegate = createRequestDelegate();
+ for (ResultSet rs : resultSets) {
+ delegate.addRequest(() -> rs.getBigDecimal(columnIndex));
+ }
+ return delegate.requestAllAndCompare();
+ }
+
+ @Override
+ public byte[] getBytes(int columnIndex) throws SQLException {
+ RequestDelegate<byte[]> delegate = createRequestDelegate();
+ for (ResultSet rs : resultSets) {
+ delegate.addRequest(() -> rs.getBytes(columnIndex));
+ }
+ return delegate.requestAllAndCompare();
+ }
+
+ @Override
+ public Date getDate(int columnIndex) throws SQLException {
+ RequestDelegate<Date> delegate = createRequestDelegate();
+ for (ResultSet rs : resultSets) {
+ delegate.addRequest(() -> rs.getDate(columnIndex));
+ }
+ return delegate.requestAllAndCompare();
+ }
+
+ @Override
+ public Time getTime(int columnIndex) throws SQLException {
+ RequestDelegate<Time> delegate = createRequestDelegate();
+ for (ResultSet rs : resultSets) {
+ delegate.addRequest(() -> rs.getTime(columnIndex));
+ }
+ return delegate.requestAllAndCompare();
+ }
+
+ @Override
+ public Timestamp getTimestamp(int columnIndex) throws SQLException {
+ RequestDelegate<Timestamp> delegate = createRequestDelegate();
+ for (ResultSet rs : resultSets) {
+ delegate.addRequest(() -> rs.getTimestamp(columnIndex));
+ }
+ return delegate.requestAllAndCompare();
+ }
+
+ @Override
+ public InputStream getAsciiStream(int columnIndex) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public InputStream getUnicodeStream(int columnIndex) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public InputStream getBinaryStream(int columnIndex) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public String getString(String columnLabel) throws SQLException {
+ RequestDelegate<String> delegate = createRequestDelegate();
+ for (ResultSet rs : resultSets) {
+ delegate.addRequest(() -> rs.getString(columnLabel));
+ }
+ return delegate.requestAllAndCompare();
+ }
+
+ @Override
+ public boolean getBoolean(String columnLabel) throws SQLException {
+ RequestDelegate<Boolean> delegate = createRequestDelegate();
+ for (ResultSet rs : resultSets) {
+ delegate.addRequest(() -> rs.getBoolean(columnLabel));
+ }
+ return delegate.requestAllAndCompare();
+ }
+
+ @Override
+ public byte getByte(String columnLabel) throws SQLException {
+ RequestDelegate<Byte> delegate = createRequestDelegate();
+ for (ResultSet rs : resultSets) {
+ delegate.addRequest(() -> rs.getByte(columnLabel));
+ }
+ return delegate.requestAllAndCompare();
+ }
+
+ @Override
+ public short getShort(String columnLabel) throws SQLException {
+ RequestDelegate<Short> delegate = createRequestDelegate();
+ for (ResultSet rs : resultSets) {
+ delegate.addRequest(() -> rs.getShort(columnLabel));
+ }
+ return delegate.requestAllAndCompare();
+ }
+
+ @Override
+ public int getInt(String columnLabel) throws SQLException {
+ RequestDelegate<Integer> delegate = createRequestDelegate();
+ for (ResultSet rs : resultSets) {
+ delegate.addRequest(() -> rs.getInt(columnLabel));
+ }
+ return delegate.requestAllAndCompare();
+ }
+
+ @Override
+ public long getLong(String columnLabel) throws SQLException {
+ RequestDelegate<Long> delegate = createRequestDelegate();
+ for (ResultSet rs : resultSets) {
+ delegate.addRequest(() -> rs.getLong(columnLabel));
+ }
+ return delegate.requestAllAndCompare();
+ }
+
+ @Override
+ public float getFloat(String columnLabel) throws SQLException {
+ RequestDelegate<Float> delegate = createRequestDelegate();
+ for (ResultSet rs : resultSets) {
+ delegate.addRequest(() -> rs.getFloat(columnLabel));
+ }
+ return delegate.requestAllAndCompare();
+ }
+
+ @Override
+ public double getDouble(String columnLabel) throws SQLException {
+ RequestDelegate<Double> delegate = createRequestDelegate();
+ for (ResultSet rs : resultSets) {
+ delegate.addRequest(() -> rs.getDouble(columnLabel));
+ }
+ return delegate.requestAllAndCompare();
+ }
+
+ @Override
+ public BigDecimal getBigDecimal(String columnLabel, int scale) throws
SQLException {
+ RequestDelegate<BigDecimal> delegate = createRequestDelegate();
+ for (ResultSet rs : resultSets) {
+ delegate.addRequest(() -> rs.getBigDecimal(columnLabel));
+ }
+ return delegate.requestAllAndCompare();
+ }
+
+ @Override
+ public byte[] getBytes(String columnLabel) throws SQLException {
+ RequestDelegate<byte[]> delegate = createRequestDelegate();
+ for (ResultSet rs : resultSets) {
+ delegate.addRequest(() -> rs.getBytes(columnLabel));
+ }
+ return delegate.requestAllAndCompare();
+ }
+
+ @Override
+ public Date getDate(String columnLabel) throws SQLException {
+ RequestDelegate<Date> delegate = createRequestDelegate();
+ for (ResultSet rs : resultSets) {
+ delegate.addRequest(() -> rs.getDate(columnLabel));
+ }
+ return delegate.requestAllAndCompare();
+ }
+
+ @Override
+ public Time getTime(String columnLabel) throws SQLException {
+ RequestDelegate<Time> delegate = createRequestDelegate();
+ for (ResultSet rs : resultSets) {
+ delegate.addRequest(() -> rs.getTime(columnLabel));
+ }
+ return delegate.requestAllAndCompare();
+ }
+
+ @Override
+ public Timestamp getTimestamp(String columnLabel) throws SQLException {
+ RequestDelegate<Timestamp> delegate = createRequestDelegate();
+ for (ResultSet rs : resultSets) {
+ delegate.addRequest(() -> rs.getTimestamp(columnLabel));
+ }
+ return delegate.requestAllAndCompare();
+ }
+
+ @Override
+ public InputStream getAsciiStream(String columnLabel) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public InputStream getUnicodeStream(String columnLabel) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public InputStream getBinaryStream(String columnLabel) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public SQLWarning getWarnings() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void clearWarnings() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public String getCursorName() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public ResultSetMetaData getMetaData() throws SQLException {
+ RequestDelegate<ResultSetMetaData> delegate = createRequestDelegate();
+ for (ResultSet rs : resultSets) {
+ delegate.addRequest(rs::getMetaData);
+ }
+ List<ResultSetMetaData> metaDataList = delegate.requestAll();
+ return new ClusterResultSetMetaData(metaDataList, endpoints);
+ }
+
+ @Override
+ public Object getObject(int columnIndex) throws SQLException {
+ RequestDelegate<Object> delegate = createRequestDelegate();
+ for (ResultSet rs : resultSets) {
+ delegate.addRequest(() -> rs.getObject(columnIndex));
+ }
+ return delegate.requestAllAndCompare();
+ }
+
+ @Override
+ public Object getObject(String columnLabel) throws SQLException {
+ RequestDelegate<Object> delegate = createRequestDelegate();
+ for (ResultSet rs : resultSets) {
+ delegate.addRequest(() -> rs.getObject(columnLabel));
+ }
+ return delegate.requestAllAndCompare();
+ }
+
+ @Override
+ public int findColumn(String columnLabel) throws SQLException {
+ RequestDelegate<Integer> delegate = createRequestDelegate();
+ for (ResultSet rs : resultSets) {
+ delegate.addRequest(() -> rs.findColumn(columnLabel));
+ }
+ return delegate.requestAllAndCompare();
+ }
+
+ @Override
+ public Reader getCharacterStream(int columnIndex) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Reader getCharacterStream(String columnLabel) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public BigDecimal getBigDecimal(int columnIndex) throws SQLException {
+ RequestDelegate<BigDecimal> delegate = createRequestDelegate();
+ for (ResultSet rs : resultSets) {
+ delegate.addRequest(() -> rs.getBigDecimal(columnIndex));
+ }
+ return delegate.requestAllAndCompare();
+ }
+
+ @Override
+ public BigDecimal getBigDecimal(String columnLabel) throws SQLException {
+ RequestDelegate<BigDecimal> delegate = createRequestDelegate();
+ for (ResultSet rs : resultSets) {
+ delegate.addRequest(() -> rs.getBigDecimal(columnLabel));
+ }
+ return delegate.requestAllAndCompare();
+ }
+
+ @Override
+ public boolean isBeforeFirst() throws SQLException {
+ RequestDelegate<Boolean> delegate = createRequestDelegate();
+ for (ResultSet rs : resultSets) {
+ delegate.addRequest(rs::isBeforeFirst);
+ }
+ return delegate.requestAllAndCompare();
+ }
+
+ @Override
+ public boolean isAfterLast() throws SQLException {
+ RequestDelegate<Boolean> delegate = createRequestDelegate();
+ for (ResultSet rs : resultSets) {
+ delegate.addRequest(rs::isAfterLast);
+ }
+ return delegate.requestAllAndCompare();
+ }
+
+ @Override
+ public boolean isFirst() throws SQLException {
+ RequestDelegate<Boolean> delegate = createRequestDelegate();
+ for (ResultSet rs : resultSets) {
+ delegate.addRequest(rs::isFirst);
+ }
+ return delegate.requestAllAndCompare();
+ }
+
+ @Override
+ public boolean isLast() throws SQLException {
+ RequestDelegate<Boolean> delegate = createRequestDelegate();
+ for (ResultSet rs : resultSets) {
+ delegate.addRequest(rs::isLast);
+ }
+ return delegate.requestAllAndCompare();
+ }
+
+ @Override
+ public void beforeFirst() throws SQLException {
+ RequestDelegate<Void> delegate = createRequestDelegate();
+ for (ResultSet rs : resultSets) {
+ delegate.addRequest(
+ () -> {
+ rs.beforeFirst();
+ return null;
+ });
+ }
+ delegate.requestAll();
+ }
+
+ @Override
+ public void afterLast() throws SQLException {
+ RequestDelegate<Void> delegate = createRequestDelegate();
+ for (ResultSet rs : resultSets) {
+ delegate.addRequest(
+ () -> {
+ rs.afterLast();
+ return null;
+ });
+ }
+ delegate.requestAll();
+ }
+
+ @Override
+ public boolean first() throws SQLException {
+ RequestDelegate<Boolean> delegate = createRequestDelegate();
+ for (ResultSet rs : resultSets) {
+ delegate.addRequest(rs::first);
+ }
+ return delegate.requestAllAndCompare();
+ }
+
+ @Override
+ public boolean last() throws SQLException {
+ RequestDelegate<Boolean> delegate = createRequestDelegate();
+ for (ResultSet rs : resultSets) {
+ delegate.addRequest(rs::last);
+ }
+ return delegate.requestAllAndCompare();
+ }
+
+ @Override
+ public int getRow() throws SQLException {
+ RequestDelegate<Integer> delegate = createRequestDelegate();
+ for (ResultSet rs : resultSets) {
+ delegate.addRequest(rs::getRow);
+ }
+ return delegate.requestAllAndCompare();
+ }
+
+ @Override
+ public boolean absolute(int row) throws SQLException {
+ RequestDelegate<Boolean> delegate = createRequestDelegate();
+ for (ResultSet rs : resultSets) {
+ delegate.addRequest(() -> rs.absolute(row));
+ }
+ return delegate.requestAllAndCompare();
+ }
+
+ @Override
+ public boolean relative(int rows) throws SQLException {
+ RequestDelegate<Boolean> delegate = createRequestDelegate();
+ for (ResultSet rs : resultSets) {
+ delegate.addRequest(() -> rs.relative(rows));
+ }
+ return delegate.requestAllAndCompare();
+ }
+
+ @Override
+ public boolean previous() throws SQLException {
+ RequestDelegate<Boolean> delegate = createRequestDelegate();
+ for (ResultSet rs : resultSets) {
+ delegate.addRequest(rs::previous);
+ }
+ return delegate.requestAllAndCompare();
+ }
+
+ @Override
+ public void setFetchDirection(int direction) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public int getFetchDirection() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void setFetchSize(int rows) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public int getFetchSize() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public int getType() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public int getConcurrency() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public boolean rowUpdated() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public boolean rowInserted() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public boolean rowDeleted() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateNull(int columnIndex) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateBoolean(int columnIndex, boolean x) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateByte(int columnIndex, byte x) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateShort(int columnIndex, short x) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateInt(int columnIndex, int x) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateLong(int columnIndex, long x) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateFloat(int columnIndex, float x) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateDouble(int columnIndex, double x) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateBigDecimal(int columnIndex, BigDecimal x) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateString(int columnIndex, String x) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateBytes(int columnIndex, byte[] x) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateDate(int columnIndex, Date x) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateTime(int columnIndex, Time x) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateTimestamp(int columnIndex, Timestamp x) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateAsciiStream(int columnIndex, InputStream x, int length) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateBinaryStream(int columnIndex, InputStream x, int length) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateCharacterStream(int columnIndex, Reader x, int length) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateObject(int columnIndex, Object x, int scaleOrLength) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateObject(int columnIndex, Object x) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateNull(String columnLabel) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateBoolean(String columnLabel, boolean x) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateByte(String columnLabel, byte x) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateShort(String columnLabel, short x) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateInt(String columnLabel, int x) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateLong(String columnLabel, long x) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateFloat(String columnLabel, float x) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateDouble(String columnLabel, double x) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateBigDecimal(String columnLabel, BigDecimal x) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateString(String columnLabel, String x) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateBytes(String columnLabel, byte[] x) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateDate(String columnLabel, Date x) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateTime(String columnLabel, Time x) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateTimestamp(String columnLabel, Timestamp x) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateAsciiStream(String columnLabel, InputStream x, int length)
{
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateBinaryStream(String columnLabel, InputStream x, int
length) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateCharacterStream(String columnLabel, Reader reader, int
length) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateObject(String columnLabel, Object x, int scaleOrLength) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateObject(String columnLabel, Object x) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void insertRow() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateRow() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void deleteRow() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void refreshRow() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void cancelRowUpdates() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void moveToInsertRow() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void moveToCurrentRow() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Statement getStatement() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Object getObject(int columnIndex, Map<String, Class<?>> map) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Ref getRef(int columnIndex) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Blob getBlob(int columnIndex) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Clob getClob(int columnIndex) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Array getArray(int columnIndex) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Object getObject(String columnLabel, Map<String, Class<?>> map) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Ref getRef(String columnLabel) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Blob getBlob(String columnLabel) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Clob getClob(String columnLabel) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Array getArray(String columnLabel) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Date getDate(int columnIndex, Calendar cal) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Date getDate(String columnLabel, Calendar cal) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Time getTime(int columnIndex, Calendar cal) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Time getTime(String columnLabel, Calendar cal) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Timestamp getTimestamp(int columnIndex, Calendar cal) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Timestamp getTimestamp(String columnLabel, Calendar cal) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public URL getURL(int columnIndex) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public URL getURL(String columnLabel) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateRef(int columnIndex, Ref x) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateRef(String columnLabel, Ref x) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateBlob(int columnIndex, Blob x) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateBlob(String columnLabel, Blob x) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateClob(int columnIndex, Clob x) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateClob(String columnLabel, Clob x) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateArray(int columnIndex, Array x) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateArray(String columnLabel, Array x) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public RowId getRowId(int columnIndex) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public RowId getRowId(String columnLabel) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateRowId(int columnIndex, RowId x) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateRowId(String columnLabel, RowId x) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public int getHoldability() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public boolean isClosed() throws SQLException {
+ RequestDelegate<Boolean> delegate = createRequestDelegate();
+ for (ResultSet rs : resultSets) {
+ delegate.addRequest(rs::isClosed);
+ }
+ return delegate.requestAllAndCompare();
+ }
+
+ @Override
+ public void updateNString(int columnIndex, String nString) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateNString(String columnLabel, String nString) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateNClob(int columnIndex, NClob nClob) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateNClob(String columnLabel, NClob nClob) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public NClob getNClob(int columnIndex) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public NClob getNClob(String columnLabel) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public SQLXML getSQLXML(int columnIndex) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public SQLXML getSQLXML(String columnLabel) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateSQLXML(int columnIndex, SQLXML xmlObject) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateSQLXML(String columnLabel, SQLXML xmlObject) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public String getNString(int columnIndex) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public String getNString(String columnLabel) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Reader getNCharacterStream(int columnIndex) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Reader getNCharacterStream(String columnLabel) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateNCharacterStream(int columnIndex, Reader x, long length) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateNCharacterStream(String columnLabel, Reader reader, long
length) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateAsciiStream(int columnIndex, InputStream x, long length) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateBinaryStream(int columnIndex, InputStream x, long length) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateCharacterStream(int columnIndex, Reader x, long length) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateAsciiStream(String columnLabel, InputStream x, long
length) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateBinaryStream(String columnLabel, InputStream x, long
length) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateCharacterStream(String columnLabel, Reader reader, long
length) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateBlob(int columnIndex, InputStream inputStream, long
length) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateBlob(String columnLabel, InputStream inputStream, long
length) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateClob(int columnIndex, Reader reader, long length) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateClob(String columnLabel, Reader reader, long length) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateNClob(int columnIndex, Reader reader, long length) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateNClob(String columnLabel, Reader reader, long length) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateNCharacterStream(int columnIndex, Reader x) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateNCharacterStream(String columnLabel, Reader reader) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateAsciiStream(int columnIndex, InputStream x) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateBinaryStream(int columnIndex, InputStream x) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateCharacterStream(int columnIndex, Reader x) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateAsciiStream(String columnLabel, InputStream x) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateBinaryStream(String columnLabel, InputStream x) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateCharacterStream(String columnLabel, Reader reader) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateBlob(int columnIndex, InputStream inputStream) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateBlob(String columnLabel, InputStream inputStream) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateClob(int columnIndex, Reader reader) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateClob(String columnLabel, Reader reader) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateNClob(int columnIndex, Reader reader) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateNClob(String columnLabel, Reader reader) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public <T> T getObject(int columnIndex, Class<T> type) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public <T> T getObject(String columnLabel, Class<T> type) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public <T> T unwrap(Class<T> iface) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public boolean isWrapperFor(Class<?> iface) {
+ throw new UnsupportedOperationException();
+ }
+
+ /**
+ * As each query operation in one ResultSet may send another RPCs to server,
the request should be
+ * executed in parallel in order to accelerate the test.
+ */
+ private <T> RequestDelegate<T> createRequestDelegate() {
+ return new ParallelRequestDelegate<>(endpoints, queryTimeoutSeconds);
+ }
+}
diff --git
a/integration-test/src/main/java/org/apache/iotdb/itbase/runtime/ClusterTestStatement.java
b/integration-test/src/main/java/org/apache/iotdb/itbase/runtime/ClusterTestStatement.java
new file mode 100644
index 0000000000..5731303cc8
--- /dev/null
+++
b/integration-test/src/main/java/org/apache/iotdb/itbase/runtime/ClusterTestStatement.java
@@ -0,0 +1,297 @@
+/*
+ * 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.itbase.runtime;
+
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.SQLWarning;
+import java.sql.Statement;
+import java.util.ArrayList;
+import java.util.List;
+
+/** The implementation of {@link ClusterTestStatement} in cluster test. */
+public class ClusterTestStatement implements Statement {
+
+ private static final int DEFAULT_QUERY_TIMEOUT = 10;
+ private final Statement writeStatement;
+ private final List<Statement> readStatements = new ArrayList<>();
+ private final List<String> readEndpoints = new ArrayList<>();
+ private boolean closed = false;
+ private int maxRows = Integer.MAX_VALUE;
+ private int queryTimeout = DEFAULT_QUERY_TIMEOUT;
+
+ public ClusterTestStatement(NodeConnection writeConnection,
List<NodeConnection> readConnections)
+ throws SQLException {
+ this.writeStatement =
writeConnection.getUnderlyingConnecton().createStatement();
+ updateConfig(writeStatement);
+ for (NodeConnection readConnection : readConnections) {
+ Statement readStatement =
readConnection.getUnderlyingConnecton().createStatement();
+ this.readStatements.add(readStatement);
+ this.readEndpoints.add(readConnection.toString());
+ updateConfig(readStatement);
+ }
+ }
+
+ private void updateConfig(Statement statement) throws SQLException {
+ maxRows = Math.min(statement.getMaxRows(), maxRows);
+ }
+
+ @Override
+ public ResultSet executeQuery(String sql) throws SQLException {
+ return new ClusterTestResultSet(readStatements, readEndpoints, sql,
queryTimeout);
+ }
+
+ @Override
+ public int executeUpdate(String sql) throws SQLException {
+ return writeStatement.executeUpdate(sql);
+ }
+
+ @Override
+ public void close() throws SQLException {
+ if (writeStatement != null) {
+ writeStatement.close();
+ }
+ readStatements.forEach(
+ r -> {
+ try {
+ r.close();
+ } catch (SQLException e) {
+ throw new RuntimeException(e);
+ }
+ });
+ closed = true;
+ }
+
+ @Override
+ public int getMaxFieldSize() throws SQLException {
+ return writeStatement.getMaxFieldSize();
+ }
+
+ @Override
+ public void setMaxFieldSize(int max) throws SQLException {
+ writeStatement.setMaxFieldSize(max);
+ }
+
+ @Override
+ public int getMaxRows() {
+ return maxRows;
+ }
+
+ @Override
+ public void setMaxRows(int max) throws SQLException {
+ for (Statement readStatement : readStatements) {
+ readStatement.setMaxRows(max);
+ }
+ maxRows = max;
+ }
+
+ @Override
+ public void setEscapeProcessing(boolean enable) throws SQLException {
+ writeStatement.setEscapeProcessing(enable);
+ for (Statement readStatement : readStatements) {
+ readStatement.setEscapeProcessing(enable);
+ }
+ }
+
+ @Override
+ public int getQueryTimeout() {
+ return queryTimeout;
+ }
+
+ @Override
+ public void setQueryTimeout(int seconds) {
+ if (seconds > 0) {
+ queryTimeout = seconds;
+ } else {
+ queryTimeout = DEFAULT_QUERY_TIMEOUT;
+ }
+ }
+
+ @Override
+ public void cancel() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public SQLWarning getWarnings() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void clearWarnings() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void setCursorName(String name) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public boolean execute(String sql) throws SQLException {
+ return writeStatement.execute(sql);
+ }
+
+ @Override
+ public ResultSet getResultSet() {
+ throw new UnsupportedOperationException(
+ "In integration-test you must use Statement.executeQuery() to query
data");
+ }
+
+ @Override
+ public int getUpdateCount() throws SQLException {
+ return writeStatement.getUpdateCount();
+ }
+
+ @Override
+ public boolean getMoreResults() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void setFetchDirection(int direction) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public int getFetchDirection() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void setFetchSize(int rows) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public int getFetchSize() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public int getResultSetConcurrency() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public int getResultSetType() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void addBatch(String sql) throws SQLException {
+ writeStatement.addBatch(sql);
+ }
+
+ @Override
+ public void clearBatch() throws SQLException {
+ writeStatement.clearBatch();
+ }
+
+ @Override
+ public int[] executeBatch() throws SQLException {
+ return writeStatement.executeBatch();
+ }
+
+ @Override
+ public Connection getConnection() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public boolean getMoreResults(int current) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public ResultSet getGeneratedKeys() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public int executeUpdate(String sql, int autoGeneratedKeys) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public int executeUpdate(String sql, int[] columnIndexes) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public int executeUpdate(String sql, String[] columnNames) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public boolean execute(String sql, int autoGeneratedKeys) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public boolean execute(String sql, int[] columnIndexes) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public boolean execute(String sql, String[] columnNames) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public int getResultSetHoldability() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public boolean isClosed() {
+ return closed;
+ }
+
+ @Override
+ public void setPoolable(boolean poolable) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public boolean isPoolable() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void closeOnCompletion() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public boolean isCloseOnCompletion() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public <T> T unwrap(Class<T> iface) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public boolean isWrapperFor(Class<?> iface) {
+ throw new UnsupportedOperationException();
+ }
+}
diff --git
a/integration-test/src/main/java/org/apache/iotdb/itbase/runtime/InconsistentDataException.java
b/integration-test/src/main/java/org/apache/iotdb/itbase/runtime/InconsistentDataException.java
new file mode 100644
index 0000000000..022f82120c
--- /dev/null
+++
b/integration-test/src/main/java/org/apache/iotdb/itbase/runtime/InconsistentDataException.java
@@ -0,0 +1,29 @@
+/*
+ * 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.itbase.runtime;
+
+import java.sql.SQLException;
+import java.util.List;
+
+public class InconsistentDataException extends SQLException {
+
+ public InconsistentDataException(Object data, List<String> endpoints) {
+ super(String.format("Datasets are inconsistent: %s with data %s",
endpoints, data));
+ }
+}
diff --git
a/integration-test/src/main/java/org/apache/iotdb/itbase/runtime/NodeConnection.java
b/integration-test/src/main/java/org/apache/iotdb/itbase/runtime/NodeConnection.java
new file mode 100644
index 0000000000..f26f9333ef
--- /dev/null
+++
b/integration-test/src/main/java/org/apache/iotdb/itbase/runtime/NodeConnection.java
@@ -0,0 +1,85 @@
+/*
+ * 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.itbase.runtime;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.sql.Connection;
+import java.sql.SQLException;
+
+/** this is the connection instance of one single node. */
+public class NodeConnection {
+
+ private static final Logger logger =
LoggerFactory.getLogger(NodeConnection.class);
+ private final String endpoint;
+ private final NodeRole nodeRole;
+ private final ConnectionRole connectionRole;
+ private final Connection underlyingConnecton;
+
+ public NodeConnection(
+ String endpoint,
+ NodeRole nodeRole,
+ ConnectionRole connectionRole,
+ Connection underlyingConnecton) {
+ this.endpoint = endpoint;
+ this.nodeRole = nodeRole;
+ this.connectionRole = connectionRole;
+ this.underlyingConnecton = underlyingConnecton;
+ }
+
+ public String getEndpoint() {
+ return endpoint;
+ }
+
+ public NodeRole getNodeRole() {
+ return nodeRole;
+ }
+
+ public ConnectionRole getConnectionRole() {
+ return connectionRole;
+ }
+
+ public Connection getUnderlyingConnecton() {
+ return underlyingConnecton;
+ }
+
+ public void close() {
+ try {
+ underlyingConnecton.close();
+ } catch (SQLException e) {
+ logger.error("Close connection {} error", this, e);
+ }
+ }
+
+ @Override
+ public String toString() {
+ return String.format("%s-%s@%s", nodeRole, connectionRole, endpoint);
+ }
+
+ public enum NodeRole {
+ CONFIG_NODE,
+ DATA_NODE,
+ }
+
+ public enum ConnectionRole {
+ READ,
+ WRITE,
+ }
+}
diff --git
a/integration-test/src/main/java/org/apache/iotdb/itbase/runtime/ParallelRequestDelegate.java
b/integration-test/src/main/java/org/apache/iotdb/itbase/runtime/ParallelRequestDelegate.java
new file mode 100644
index 0000000000..32d9803032
--- /dev/null
+++
b/integration-test/src/main/java/org/apache/iotdb/itbase/runtime/ParallelRequestDelegate.java
@@ -0,0 +1,60 @@
+/*
+ * 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.itbase.runtime;
+
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * ParallelRequestDelegate will handle requests in parallel. It's more
efficient when requests
+ * contain network communication.
+ */
+public class ParallelRequestDelegate<T> extends RequestDelegate<T> {
+ private final int taskTimeoutSeconds;
+
+ public ParallelRequestDelegate(List<String> endpoints, int
taskTimeoutSeconds) {
+ super(endpoints);
+ this.taskTimeoutSeconds = taskTimeoutSeconds;
+ }
+
+ public List<T> requestAll() throws SQLException {
+ List<Future<T>> resultFutures = new ArrayList<>(getRequests().size());
+ for (Callable<T> request : getRequests()) {
+ Future<T> f = RequestThreadPool.submit(request);
+ resultFutures.add(f);
+ }
+ List<T> results = new ArrayList<>(getRequests().size());
+ for (int i = 0; i < getEndpoints().size(); i++) {
+ try {
+ results.add(resultFutures.get(i).get(taskTimeoutSeconds,
TimeUnit.SECONDS));
+ } catch (Exception e) {
+ for (int j = i + 1; j < getEndpoints().size(); j++) {
+ resultFutures.get(j).cancel(true);
+ }
+ throw new SQLException(
+ String.format("Waiting for query results of %s failed",
getEndpoints().get(i)), e);
+ }
+ }
+ return results;
+ }
+}
diff --git
a/integration-test/src/main/java/org/apache/iotdb/itbase/runtime/RequestDelegate.java
b/integration-test/src/main/java/org/apache/iotdb/itbase/runtime/RequestDelegate.java
new file mode 100644
index 0000000000..af71fe1577
--- /dev/null
+++
b/integration-test/src/main/java/org/apache/iotdb/itbase/runtime/RequestDelegate.java
@@ -0,0 +1,86 @@
+/*
+ * 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.itbase.runtime;
+
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.Callable;
+
+/** This class is used to handle multi requests and gather their returned
values. */
+public abstract class RequestDelegate<T> {
+
+ private final List<String> endpoints;
+ private final List<Callable<T>> requests = new ArrayList<>();
+
+ /**
+ * Initialize a RequestDelegate instance with a list of endpoints.
+ *
+ * @param endpoints the list of endpoints.
+ */
+ public RequestDelegate(List<String> endpoints) {
+ this.endpoints = endpoints;
+ }
+
+ /**
+ * Add a request to this RequestDelegate. The request sequence should
correspond to endpoints.
+ *
+ * @param request the request to add.
+ */
+ public final void addRequest(Callable<T> request) {
+ requests.add(request);
+ }
+
+ /**
+ * Do the requests which have been added, and return a list of their return
values.
+ *
+ * @return the return values of all the request added in order.
+ * @throws SQLException if any error happens during requesting.
+ */
+ public abstract List<T> requestAll() throws SQLException;
+
+ /**
+ * Do the requests which have been added, and then comparing their return
values. If all the
+ * values are equal, one value will be returned. Otherwise an {@link
SQLException} will be thrown.
+ *
+ * @return the return value of one request, as all requests return the same
one.
+ * @throws SQLException if any error happens during requesting.
+ * @throws InconsistentDataException if the return values of requests are
not equal.
+ */
+ public final T requestAllAndCompare() throws SQLException {
+ List<T> results = requestAll();
+ T data = results.get(0);
+ for (int i = 1; i < results.size(); i++) {
+ T anotherData = results.get(i);
+ if (!Objects.equals(data, anotherData)) {
+ throw new InconsistentDataException(results, endpoints);
+ }
+ }
+ return data;
+ }
+
+ protected List<String> getEndpoints() {
+ return endpoints;
+ }
+
+ protected List<Callable<T>> getRequests() {
+ return requests;
+ }
+}
diff --git
a/integration-test/src/main/java/org/apache/iotdb/itbase/runtime/RequestThreadPool.java
b/integration-test/src/main/java/org/apache/iotdb/itbase/runtime/RequestThreadPool.java
new file mode 100644
index 0000000000..9ff1c98b07
--- /dev/null
+++
b/integration-test/src/main/java/org/apache/iotdb/itbase/runtime/RequestThreadPool.java
@@ -0,0 +1,49 @@
+/*
+ * 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.itbase.runtime;
+
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+
+/**
+ * A singleton thread pool used for handling test requests in one single JVM.
This can be used to
+ * avoid thread leak and frequency thread allocation.
+ */
+public class RequestThreadPool {
+ private final ExecutorService executor;
+
+ private RequestThreadPool() {
+ executor = Executors.newFixedThreadPool(5);
+ }
+
+ public static <T> Future<T> submit(Callable<T> task) {
+ return InstanceHolder.INSTANCE.executor.submit(task);
+ }
+
+ private static class InstanceHolder {
+
+ private InstanceHolder() {
+ // nothing to do
+ }
+
+ private static final RequestThreadPool INSTANCE = new RequestThreadPool();
+ }
+}
diff --git
a/integration-test/src/main/java/org/apache/iotdb/itbase/runtime/SerialRequestDelegate.java
b/integration-test/src/main/java/org/apache/iotdb/itbase/runtime/SerialRequestDelegate.java
new file mode 100644
index 0000000000..10e53b62e6
--- /dev/null
+++
b/integration-test/src/main/java/org/apache/iotdb/itbase/runtime/SerialRequestDelegate.java
@@ -0,0 +1,47 @@
+/*
+ * 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.itbase.runtime;
+
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * ParallelRequestDelegate will handle requests in serial. It's more efficient
when the requests are
+ * just local computation.
+ */
+public class SerialRequestDelegate<T> extends RequestDelegate<T> {
+
+ public SerialRequestDelegate(List<String> endpoints) {
+ super(endpoints);
+ }
+
+ @Override
+ public List<T> requestAll() throws SQLException {
+ List<T> results = new ArrayList<>(getEndpoints().size());
+ for (int i = 0; i < getEndpoints().size(); i++) {
+ try {
+ results.add(getRequests().get(i).call());
+ } catch (Exception e) {
+ throw new SQLException(String.format("Request %s error.",
getEndpoints().get(i)), e);
+ }
+ }
+ return results;
+ }
+}
diff --git a/metrics/pom.xml b/metrics/pom.xml
index 1e0aba9503..ec35d6f971 100644
--- a/metrics/pom.xml
+++ b/metrics/pom.xml
@@ -32,9 +32,50 @@
<name>metric module</name>
<description>Metrics interface for IoTDB</description>
<url>https://github.com/thulab/iotdb/tree/master/tsfile</url>
+ <properties>
+ <metrics.test.skip>false</metrics.test.skip>
+ </properties>
<modules>
<module>interface</module>
<module>micrometer-metrics</module>
<module>dropwizard-metrics</module>
</modules>
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-surefire-plugin</artifactId>
+ <configuration>
+ <skipTests>${metrics.test.skip}</skipTests>
+ <forkMode>pertest</forkMode>
+ </configuration>
+ </plugin>
+ </plugins>
+ </build>
+ <profiles>
+ <profile>
+ <id>skipMetricsTests</id>
+ <activation>
+ <property>
+ <name>skipTests</name>
+ <value>true</value>
+ </property>
+ </activation>
+ <properties>
+ <metrics.test.skip>true</metrics.test.skip>
+ </properties>
+ </profile>
+ <profile>
+ <id>skipUT_Metrics_Tests</id>
+ <activation>
+ <property>
+ <name>skipUTs</name>
+ <value>true</value>
+ </property>
+ </activation>
+ <properties>
+ <metrics.test.skip>true</metrics.test.skip>
+ </properties>
+ </profile>
+ </profiles>
</project>
