You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ro...@apache.org on 2024/11/14 07:29:53 UTC
(iotdb) branch master updated: Pipe IT: Add table model related IT (#14086)
This is an automated email from the ASF dual-hosted git repository.
rong 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 86a33773e8c Pipe IT: Add table model related IT (#14086)
86a33773e8c is described below
commit 86a33773e8c95be59837fb7704967ec4b0cdc54b
Author: Zhenyu Luo <lu...@gmail.com>
AuthorDate: Thu Nov 14 15:29:47 2024 +0800
Pipe IT: Add table model related IT (#14086)
---
.github/workflows/pipe-it-2cluster.yml | 37 +
integration-test/pom.xml | 16 +-
.../iotdb/it/env/cluster/env/AbstractEnv.java | 11 +-
.../iotdb/it/env/remote/env/RemoteServerEnv.java | 2 +-
.../itbase/category/MultiClusterIT2TableModel.java | 22 +
.../java/org/apache/iotdb/itbase/env/BaseEnv.java | 11 +-
.../org/apache/iotdb/db/it/utils/TestUtils.java | 141 +++-
.../tablemodel/AbstractPipeTableModelTestIT.java | 69 ++
.../iotdb/pipe/it/tablemodel/IoTDBPipeAlterIT.java | 457 ++++++++++
.../it/tablemodel/IoTDBPipeAutoConflictIT.java | 267 ++++++
.../pipe/it/tablemodel/IoTDBPipeAutoDropIT.java | 152 ++++
.../pipe/it/tablemodel/IoTDBPipeClusterIT.java | 909 ++++++++++++++++++++
.../IoTDBPipeConnectorCompressionIT.java | 326 ++++++++
.../tablemodel/IoTDBPipeConnectorParallelIT.java | 99 +++
.../pipe/it/tablemodel/IoTDBPipeDataSinkIT.java | 219 +++++
.../pipe/it/tablemodel/IoTDBPipeExtractorIT.java | 921 +++++++++++++++++++++
.../pipe/it/tablemodel/IoTDBPipeLifeCycleIT.java | 642 ++++++++++++++
.../pipe/it/tablemodel/IoTDBPipeNullValueIT.java | 169 ++++
.../pipe/it/tablemodel/IoTDBPipeProtocolIT.java | 466 +++++++++++
.../it/tablemodel/IoTDBPipeSwitchStatusIT.java | 348 ++++++++
.../pipe/it/tablemodel/IoTDBPipeSyntaxIT.java | 685 +++++++++++++++
.../pipe/it/tablemodel/IoTDBPipeWithLoadIT.java | 120 +++
.../it/tablemodel/IoTDBTablePatternFormatIT.java | 572 +++++++++++++
.../iotdb/pipe/it/tablemodel/TableModelUtils.java | 395 +++++++++
24 files changed, 7048 insertions(+), 8 deletions(-)
diff --git a/.github/workflows/pipe-it-2cluster.yml b/.github/workflows/pipe-it-2cluster.yml
index 29aaeea3fc6..aca219cdcbd 100644
--- a/.github/workflows/pipe-it-2cluster.yml
+++ b/.github/workflows/pipe-it-2cluster.yml
@@ -234,3 +234,40 @@ jobs:
name: cluster-log-subscription-regression-misc-java${{ matrix.java }}-${{ runner.os }}-${{ matrix.cluster1 }}-${{ matrix.cluster2 }}
path: integration-test/target/cluster-logs
retention-days: 30
+ table-model:
+ strategy:
+ fail-fast: false
+ max-parallel: 15
+ matrix:
+ java: [17]
+ # StrongConsistencyClusterMode is ignored now because RatisConsensus has not been supported yet.
+ cluster: [LightWeightStandaloneMode, ScalableSingleNodeMode, HighPerformanceMode]
+ os: [ ubuntu-latest ]
+ runs-on: ${{ matrix.os }}
+ steps:
+ - uses: actions/checkout@v4
+ - name: Set up JDK ${{ matrix.java }}
+ uses: actions/setup-java@v4
+ with:
+ distribution: liberica
+ java-version: ${{ matrix.java }}
+ - name: IT Test
+ 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 clean verify \
+ -P with-integration-tests \
+ -DskipUTs \
+ -DintegrationTest.forkCount=1 -DConfigNodeMaxHeapSize=256 -DDataNodeMaxHeapSize=1024 -DDataNodeMaxDirectMemorySize=768 \
+ -DClusterConfigurations=${{ matrix.cluster }},${{ matrix.cluster }} \
+ -pl integration-test \
+ -am -PMultiClusterIT2TableModel \
+ -ntp
+ - name: Upload Artifact
+ if: failure()
+ uses: actions/upload-artifact@v4
+ with:
+ name: cluster-log-table-model-java${{ matrix.java }}-${{ runner.os }}-${{ matrix.cluster }}-${{ matrix.cluster }}
+ path: integration-test/target/cluster-logs
+ retention-days: 30
diff --git a/integration-test/pom.xml b/integration-test/pom.xml
index da5f9b7c1c0..75fd871993d 100644
--- a/integration-test/pom.xml
+++ b/integration-test/pom.xml
@@ -415,7 +415,7 @@
</activation>
<properties>
<integrationTest.excludedGroups/>
- <integrationTest.includedGroups>org.apache.iotdb.itbase.category.MultiClusterIT1,org.apache.iotdb.itbase.category.MultiClusterIT2AutoCreateSchema,org.apache.iotdb.itbase.category.MultiClusterIT2ManualCreateSchema,org.apache.iotdb.itbase.category.MultiClusterIT2SubscriptionArchVerification,org.apache.iotdb.itbase.category.MultiClusterIT2SubscriptionRegressionConsumer,org.apache.iotdb.itbase.category.MultiClusterIT2SubscriptionRegressionMisc,org.apache.iotdb.itbase.category [...]
+ <integrationTest.includedGroups>org.apache.iotdb.itbase.category.MultiClusterIT1,org.apache.iotdb.itbase.category.MultiClusterIT2AutoCreateSchema,org.apache.iotdb.itbase.category.MultiClusterIT2ManualCreateSchema,org.apache.iotdb.itbase.category.MultiClusterIT2SubscriptionArchVerification,org.apache.iotdb.itbase.category.MultiClusterIT2SubscriptionRegressionConsumer,org.apache.iotdb.itbase.category.MultiClusterIT2SubscriptionRegressionMisc,org.apache.iotdb.itbase.category [...]
<integrationTest.launchNodeInSameJVM>false</integrationTest.launchNodeInSameJVM>
<integrationTest.randomSelectWriteNode>true</integrationTest.randomSelectWriteNode>
<integrationTest.readAndVerifyWithMultiNode>true</integrationTest.readAndVerifyWithMultiNode>
@@ -464,6 +464,20 @@
<integrationTest.testEnv>MultiCluster</integrationTest.testEnv>
</properties>
</profile>
+ <profile>
+ <id>MultiClusterIT2TableModel</id>
+ <activation>
+ <activeByDefault>false</activeByDefault>
+ </activation>
+ <properties>
+ <integrationTest.excludedGroups/>
+ <integrationTest.includedGroups>org.apache.iotdb.itbase.category.MultiClusterIT2TableModel</integrationTest.includedGroups>
+ <integrationTest.launchNodeInSameJVM>false</integrationTest.launchNodeInSameJVM>
+ <integrationTest.randomSelectWriteNode>true</integrationTest.randomSelectWriteNode>
+ <integrationTest.readAndVerifyWithMultiNode>true</integrationTest.readAndVerifyWithMultiNode>
+ <integrationTest.testEnv>MultiCluster</integrationTest.testEnv>
+ </properties>
+ </profile>
<profile>
<id>MultiClusterIT2SubscriptionArchVerification</id>
<activation>
diff --git a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/env/AbstractEnv.java b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/env/AbstractEnv.java
index 2b2a262dbd2..90d61d754c6 100644
--- a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/env/AbstractEnv.java
+++ b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/env/AbstractEnv.java
@@ -436,11 +436,18 @@ public abstract class AbstractEnv implements BaseEnv {
@Override
public Connection getWriteOnlyConnectionWithSpecifiedDataNode(
- final DataNodeWrapper dataNode, final String username, final String password)
+ final DataNodeWrapper dataNode,
+ final String username,
+ final String password,
+ String sqlDialect)
throws SQLException {
return new ClusterTestConnection(
getWriteConnectionWithSpecifiedDataNode(
- dataNode, null, username, password, TREE_SQL_DIALECT),
+ dataNode,
+ null,
+ username,
+ password,
+ TABLE_SQL_DIALECT.equals(sqlDialect) ? TABLE_SQL_DIALECT : TREE_SQL_DIALECT),
Collections.emptyList());
}
diff --git a/integration-test/src/main/java/org/apache/iotdb/it/env/remote/env/RemoteServerEnv.java b/integration-test/src/main/java/org/apache/iotdb/it/env/remote/env/RemoteServerEnv.java
index 04e3399ad32..70617f41f28 100644
--- a/integration-test/src/main/java/org/apache/iotdb/it/env/remote/env/RemoteServerEnv.java
+++ b/integration-test/src/main/java/org/apache/iotdb/it/env/remote/env/RemoteServerEnv.java
@@ -137,7 +137,7 @@ public class RemoteServerEnv implements BaseEnv {
@Override
public Connection getWriteOnlyConnectionWithSpecifiedDataNode(
- DataNodeWrapper dataNode, String username, String password) {
+ DataNodeWrapper dataNode, String username, String password, String sqlDialect) {
throw new UnsupportedOperationException();
}
diff --git a/integration-test/src/main/java/org/apache/iotdb/itbase/category/MultiClusterIT2TableModel.java b/integration-test/src/main/java/org/apache/iotdb/itbase/category/MultiClusterIT2TableModel.java
new file mode 100644
index 00000000000..c9280cd3fc6
--- /dev/null
+++ b/integration-test/src/main/java/org/apache/iotdb/itbase/category/MultiClusterIT2TableModel.java
@@ -0,0 +1,22 @@
+/*
+ * 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.category;
+
+public class MultiClusterIT2TableModel {}
diff --git a/integration-test/src/main/java/org/apache/iotdb/itbase/env/BaseEnv.java b/integration-test/src/main/java/org/apache/iotdb/itbase/env/BaseEnv.java
index 1d721aaa798..4ef14786d8c 100644
--- a/integration-test/src/main/java/org/apache/iotdb/itbase/env/BaseEnv.java
+++ b/integration-test/src/main/java/org/apache/iotdb/itbase/env/BaseEnv.java
@@ -147,12 +147,19 @@ public interface BaseEnv {
default Connection getWriteOnlyConnectionWithSpecifiedDataNode(DataNodeWrapper dataNode)
throws SQLException {
return getWriteOnlyConnectionWithSpecifiedDataNode(
- dataNode, SessionConfig.DEFAULT_USER, SessionConfig.DEFAULT_PASSWORD);
+ dataNode, SessionConfig.DEFAULT_USER, SessionConfig.DEFAULT_PASSWORD, TREE_SQL_DIALECT);
+ }
+
+ default Connection getWriteOnlyConnectionWithSpecifiedDataNode(
+ DataNodeWrapper dataNode, String sqlDialect) throws SQLException {
+ return getWriteOnlyConnectionWithSpecifiedDataNode(
+ dataNode, SessionConfig.DEFAULT_USER, SessionConfig.DEFAULT_PASSWORD, sqlDialect);
}
// This is useful when you shut down a dataNode.
Connection getWriteOnlyConnectionWithSpecifiedDataNode(
- DataNodeWrapper dataNode, String username, String password) throws SQLException;
+ DataNodeWrapper dataNode, String username, String password, String sqlDialect)
+ throws SQLException;
default Connection getConnectionWithSpecifiedDataNode(DataNodeWrapper dataNode)
throws SQLException {
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 641c8c31cc2..6e201567cd9 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
@@ -638,24 +638,82 @@ public class TestUtils {
env, sql, SessionConfig.DEFAULT_USER, SessionConfig.DEFAULT_PASSWORD);
}
+ public static boolean tryExecuteNonQueryWithRetry(
+ String dataBaseName, String sqlDialect, BaseEnv env, String sql) {
+ return tryExecuteNonQueryWithRetry(
+ env,
+ sql,
+ SessionConfig.DEFAULT_USER,
+ SessionConfig.DEFAULT_PASSWORD,
+ dataBaseName,
+ sqlDialect);
+ }
+
public static boolean tryExecuteNonQueryWithRetry(
BaseEnv env, String sql, String userName, String password) {
return tryExecuteNonQueriesWithRetry(env, Collections.singletonList(sql), userName, password);
}
+ public static boolean tryExecuteNonQueryWithRetry(
+ BaseEnv env,
+ String sql,
+ String userName,
+ String password,
+ String dataBaseName,
+ String sqlDialect) {
+ return tryExecuteNonQueriesWithRetry(
+ env, Collections.singletonList(sql), userName, password, dataBaseName, sqlDialect);
+ }
+
public static boolean tryExecuteNonQueriesWithRetry(BaseEnv env, List<String> sqlList) {
return tryExecuteNonQueriesWithRetry(
- env, sqlList, SessionConfig.DEFAULT_USER, SessionConfig.DEFAULT_PASSWORD);
+ env,
+ sqlList,
+ SessionConfig.DEFAULT_USER,
+ SessionConfig.DEFAULT_PASSWORD,
+ null,
+ BaseEnv.TREE_SQL_DIALECT);
+ }
+
+ public static boolean tryExecuteNonQueriesWithRetry(
+ String dataBase, String sqlDialect, BaseEnv env, List<String> sqlList) {
+ return tryExecuteNonQueriesWithRetry(
+ env,
+ sqlList,
+ SessionConfig.DEFAULT_USER,
+ SessionConfig.DEFAULT_PASSWORD,
+ dataBase,
+ sqlDialect);
}
// This method will not throw failure given that a failure is encountered.
// Instead, it returns a flag to indicate the result of the execution.
public static boolean tryExecuteNonQueriesWithRetry(
BaseEnv env, List<String> sqlList, String userName, String password) {
+ return tryExecuteNonQueriesWithRetry(
+ env, sqlList, userName, password, null, BaseEnv.TREE_SQL_DIALECT);
+ }
+
+ public static boolean tryExecuteNonQueriesWithRetry(
+ BaseEnv env,
+ List<String> sqlList,
+ String userName,
+ String password,
+ String dataBase,
+ String sqlDialect) {
int lastIndex = 0;
for (int retryCountLeft = 10; retryCountLeft >= 0; retryCountLeft--) {
- try (Connection connection = env.getConnection(userName, password);
+ try (Connection connection =
+ env.getConnection(
+ userName,
+ password,
+ BaseEnv.TABLE_SQL_DIALECT.equals(sqlDialect)
+ ? BaseEnv.TABLE_SQL_DIALECT
+ : BaseEnv.TREE_SQL_DIALECT);
Statement statement = connection.createStatement()) {
+ if (BaseEnv.TABLE_SQL_DIALECT.equals(sqlDialect) && dataBase != null) {
+ statement.execute("use " + dataBase);
+ }
for (int i = lastIndex; i < sqlList.size(); ++i) {
lastIndex = i;
statement.execute(sqlList.get(i));
@@ -731,6 +789,42 @@ public class TestUtils {
return false;
}
+ public static boolean tryExecuteNonQueriesOnSpecifiedDataNodeWithRetry(
+ BaseEnv env,
+ DataNodeWrapper wrapper,
+ List<String> sqlList,
+ String dataBase,
+ String sqlDialect) {
+ int lastIndex = 0;
+ for (int retryCountLeft = 10; retryCountLeft >= 0; retryCountLeft--) {
+ try (Connection connection =
+ env.getWriteOnlyConnectionWithSpecifiedDataNode(wrapper, sqlDialect);
+ Statement statement = connection.createStatement()) {
+
+ if (BaseEnv.TABLE_SQL_DIALECT.equals(sqlDialect) && dataBase != null) {
+ statement.execute("use " + dataBase);
+ }
+
+ for (int i = lastIndex; i < sqlList.size(); ++i) {
+ statement.execute(sqlList.get(i));
+ lastIndex = i;
+ }
+ return true;
+ } catch (SQLException e) {
+ if (retryCountLeft > 0) {
+ try {
+ Thread.sleep(10000);
+ } catch (InterruptedException ignored) {
+ }
+ } else {
+ e.printStackTrace();
+ return false;
+ }
+ }
+ }
+ return false;
+ }
+
public static void executeQuery(String sql) {
executeQuery(sql, SessionConfig.DEFAULT_USER, SessionConfig.DEFAULT_PASSWORD);
}
@@ -909,6 +1003,49 @@ public class TestUtils {
}
}
+ public static void assertDataEventuallyOnEnv(
+ BaseEnv env,
+ String sql,
+ String expectedHeader,
+ Set<String> expectedResSet,
+ String dataBaseName) {
+ assertDataEventuallyOnEnv(env, sql, expectedHeader, expectedResSet, 300, dataBaseName);
+ }
+
+ public static void assertDataEventuallyOnEnv(
+ BaseEnv env,
+ String sql,
+ String expectedHeader,
+ Set<String> expectedResSet,
+ long timeoutSeconds,
+ String dataBaseName) {
+ try (Connection connection = env.getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ Statement statement = connection.createStatement()) {
+ // Keep retrying if there are execution failures
+ await()
+ .pollInSameThread()
+ .pollDelay(1L, TimeUnit.SECONDS)
+ .pollInterval(1L, TimeUnit.SECONDS)
+ .atMost(timeoutSeconds, TimeUnit.SECONDS)
+ .untilAsserted(
+ () -> {
+ try {
+ if (dataBaseName != null) {
+ statement.execute("use " + dataBaseName);
+ }
+ if (sql != null && !sql.equals("")) {
+ TestUtils.assertResultSetEqual(
+ executeQueryWithRetry(statement, sql), expectedHeader, expectedResSet);
+ }
+ } catch (Exception e) {
+ Assert.fail();
+ }
+ });
+ } catch (Exception e) {
+ fail(e.getMessage());
+ }
+ }
+
public static void assertDataEventuallyOnEnv(
BaseEnv env, String sql, Map<String, String> expectedHeaderWithResult) {
assertDataEventuallyOnEnv(env, sql, expectedHeaderWithResult, 600);
diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/tablemodel/AbstractPipeTableModelTestIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/tablemodel/AbstractPipeTableModelTestIT.java
new file mode 100644
index 00000000000..b18ac32aae5
--- /dev/null
+++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/tablemodel/AbstractPipeTableModelTestIT.java
@@ -0,0 +1,69 @@
+/*
+ * 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.pipe.it.tablemodel;
+
+import org.apache.iotdb.consensus.ConsensusFactory;
+import org.apache.iotdb.it.env.MultiEnvFactory;
+import org.apache.iotdb.itbase.env.BaseEnv;
+
+import org.junit.After;
+import org.junit.Before;
+
+abstract class AbstractPipeTableModelTestIT {
+
+ protected BaseEnv senderEnv;
+ protected BaseEnv receiverEnv;
+
+ @Before
+ public void setUp() {
+ MultiEnvFactory.createEnv(2);
+ senderEnv = MultiEnvFactory.getEnv(0);
+ receiverEnv = MultiEnvFactory.getEnv(1);
+ setupConfig();
+ senderEnv.initClusterEnvironment();
+ receiverEnv.initClusterEnvironment();
+ }
+
+ protected void setupConfig() {
+ // TODO: delete ratis configurations
+ senderEnv
+ .getConfig()
+ .getCommonConfig()
+ .setAutoCreateSchemaEnabled(true)
+ .setConfigNodeConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS)
+ .setSchemaRegionConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS);
+ receiverEnv
+ .getConfig()
+ .getCommonConfig()
+ .setAutoCreateSchemaEnabled(true)
+ .setConfigNodeConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS)
+ .setSchemaRegionConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS);
+
+ // 10 min, assert that the operations will not time out
+ senderEnv.getConfig().getCommonConfig().setCnConnectionTimeoutMs(600000);
+ receiverEnv.getConfig().getCommonConfig().setCnConnectionTimeoutMs(600000);
+ }
+
+ @After
+ public final void tearDown() {
+ senderEnv.cleanClusterEnvironment();
+ receiverEnv.cleanClusterEnvironment();
+ }
+}
diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/tablemodel/IoTDBPipeAlterIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/tablemodel/IoTDBPipeAlterIT.java
new file mode 100644
index 00000000000..0f682c7f666
--- /dev/null
+++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/tablemodel/IoTDBPipeAlterIT.java
@@ -0,0 +1,457 @@
+/*
+ * 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.pipe.it.tablemodel;
+
+import org.apache.iotdb.commons.client.sync.SyncConfigNodeIServiceClient;
+import org.apache.iotdb.confignode.rpc.thrift.TShowPipeInfo;
+import org.apache.iotdb.confignode.rpc.thrift.TShowPipeReq;
+import org.apache.iotdb.db.it.utils.TestUtils;
+import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper;
+import org.apache.iotdb.it.framework.IoTDBTestRunner;
+import org.apache.iotdb.itbase.category.MultiClusterIT2TableModel;
+import org.apache.iotdb.itbase.env.BaseEnv;
+
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import java.sql.Connection;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.List;
+
+import static org.junit.Assert.fail;
+
+@RunWith(IoTDBTestRunner.class)
+@Category({MultiClusterIT2TableModel.class})
+public class IoTDBPipeAlterIT extends AbstractPipeTableModelTestIT {
+
+ @Test
+ public void testBasicAlterPipe() throws Exception {
+ final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+
+ // Create pipe
+ final String sql =
+ String.format(
+ "create pipe a2b with source ('source'='iotdb-source', 'database-name'='test', 'table-name'='test1', 'mode.streaming'='true') with processor ('processor'='do-nothing-processor') with sink ('node-urls'='%s')",
+ receiverDataNode.getIpAndPortString());
+ try (final Connection connection = senderEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ final Statement statement = connection.createStatement()) {
+ statement.execute(sql);
+ } catch (SQLException e) {
+ fail(e.getMessage());
+ }
+
+ // show pipe
+ long lastCreationTime;
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+ final List<TShowPipeInfo> showPipeResult = client.showPipe(new TShowPipeReq()).pipeInfoList;
+ Assert.assertEquals(1, showPipeResult.size());
+ // Check status
+ Assert.assertEquals("RUNNING", showPipeResult.get(0).state);
+ // Check configurations
+ Assert.assertTrue(showPipeResult.get(0).pipeExtractor.contains("source=iotdb-source"));
+ Assert.assertTrue(showPipeResult.get(0).pipeExtractor.contains("database-name=test"));
+ Assert.assertTrue(showPipeResult.get(0).pipeExtractor.contains("table-name=test"));
+ Assert.assertTrue(showPipeResult.get(0).pipeExtractor.contains("mode.streaming=true"));
+ Assert.assertTrue(
+ showPipeResult.get(0).pipeProcessor.contains("processor=do-nothing-processor"));
+ Assert.assertTrue(
+ showPipeResult
+ .get(0)
+ .pipeConnector
+ .contains(String.format("node-urls=%s", receiverDataNode.getIpAndPortString())));
+ // Record last creation time
+ lastCreationTime = showPipeResult.get(0).creationTime;
+ }
+
+ // Stop pipe
+ try (final Connection connection = senderEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ final Statement statement = connection.createStatement()) {
+ statement.execute("stop pipe a2b");
+ } catch (SQLException e) {
+ fail(e.getMessage());
+ }
+
+ // Show pipe
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+ final List<TShowPipeInfo> showPipeResult = client.showPipe(new TShowPipeReq()).pipeInfoList;
+ Assert.assertEquals(1, showPipeResult.size());
+ // Check status
+ Assert.assertEquals("STOPPED", showPipeResult.get(0).state);
+ }
+
+ // Alter pipe (modify)
+ try (final Connection connection = senderEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ final Statement statement = connection.createStatement()) {
+ statement.execute(
+ "alter pipe a2b modify source ('table-name'='test1','database-name'='test1')");
+ } catch (SQLException e) {
+ fail(e.getMessage());
+ }
+
+ // Show pipe
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+ final List<TShowPipeInfo> showPipeResult = client.showPipe(new TShowPipeReq()).pipeInfoList;
+ Assert.assertEquals(1, showPipeResult.size());
+ // Check status
+ Assert.assertEquals("STOPPED", showPipeResult.get(0).state);
+ // Check configurations
+ Assert.assertTrue(showPipeResult.get(0).pipeExtractor.contains("source=iotdb-source"));
+ Assert.assertTrue(showPipeResult.get(0).pipeExtractor.contains("table-name=test1"));
+ Assert.assertTrue(showPipeResult.get(0).pipeExtractor.contains("database-name=test1"));
+ Assert.assertTrue(showPipeResult.get(0).pipeExtractor.contains("mode.streaming=true"));
+ Assert.assertTrue(
+ showPipeResult.get(0).pipeProcessor.contains("processor=do-nothing-processor"));
+ Assert.assertTrue(
+ showPipeResult
+ .get(0)
+ .pipeConnector
+ .contains(String.format("node-urls=%s", receiverDataNode.getIpAndPortString())));
+ // Check creation time and record last creation time
+ Assert.assertTrue(showPipeResult.get(0).creationTime > lastCreationTime);
+ lastCreationTime = showPipeResult.get(0).creationTime;
+ // Check exception message
+ Assert.assertEquals("", showPipeResult.get(0).exceptionMessage);
+ }
+
+ // Alter pipe (replace)
+ try (final Connection connection = senderEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ final Statement statement = connection.createStatement()) {
+ statement.execute(
+ "alter pipe a2b replace source ('capture.table'='true','''source'='iotdb-source', 'table-name'='test','database-name'='test')");
+ } catch (SQLException e) {
+ fail(e.getMessage());
+ }
+
+ // Show pipe
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+ final List<TShowPipeInfo> showPipeResult = client.showPipe(new TShowPipeReq()).pipeInfoList;
+ Assert.assertEquals(1, showPipeResult.size());
+ // check status
+ Assert.assertEquals("STOPPED", showPipeResult.get(0).state);
+ // check configurations
+ Assert.assertTrue(showPipeResult.get(0).pipeExtractor.contains("source=iotdb-source"));
+ Assert.assertTrue(showPipeResult.get(0).pipeExtractor.contains("database-name=test"));
+ Assert.assertTrue(showPipeResult.get(0).pipeExtractor.contains("table-name=test"));
+ Assert.assertFalse(showPipeResult.get(0).pipeExtractor.contains("mode.streaming=true"));
+ Assert.assertTrue(
+ showPipeResult.get(0).pipeProcessor.contains("processor=do-nothing-processor"));
+ Assert.assertTrue(
+ showPipeResult
+ .get(0)
+ .pipeConnector
+ .contains(String.format("node-urls=%s", receiverDataNode.getIpAndPortString())));
+ // Check creation time and record last creation time
+ Assert.assertTrue(showPipeResult.get(0).creationTime > lastCreationTime);
+ lastCreationTime = showPipeResult.get(0).creationTime;
+ // check exception message
+ Assert.assertEquals("", showPipeResult.get(0).exceptionMessage);
+ }
+
+ // Alter pipe (modify)
+ try (final Connection connection = senderEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ final Statement statement = connection.createStatement()) {
+ statement.execute("alter pipe a2b modify sink ('sink.batch.enable'='false')");
+ } catch (SQLException e) {
+ fail(e.getMessage());
+ }
+
+ // Show pipe
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+ final List<TShowPipeInfo> showPipeResult = client.showPipe(new TShowPipeReq()).pipeInfoList;
+ Assert.assertEquals(1, showPipeResult.size());
+ // Check status
+ Assert.assertEquals("STOPPED", showPipeResult.get(0).state);
+ // Check configurations
+ Assert.assertTrue(showPipeResult.get(0).pipeExtractor.contains("source=iotdb-source"));
+ Assert.assertTrue(showPipeResult.get(0).pipeExtractor.contains("database-name=test"));
+ Assert.assertTrue(showPipeResult.get(0).pipeExtractor.contains("table-name=test"));
+ Assert.assertTrue(showPipeResult.get(0).pipeConnector.contains("batch.enable=false"));
+ Assert.assertTrue(
+ showPipeResult.get(0).pipeProcessor.contains("processor=do-nothing-processor"));
+ Assert.assertTrue(
+ showPipeResult
+ .get(0)
+ .pipeConnector
+ .contains(String.format("node-urls=%s", receiverDataNode.getIpAndPortString())));
+ // Check creation time and record last creation time
+ Assert.assertTrue(showPipeResult.get(0).creationTime > lastCreationTime);
+ lastCreationTime = showPipeResult.get(0).creationTime;
+ // Check exception message
+ Assert.assertEquals("", showPipeResult.get(0).exceptionMessage);
+ }
+
+ // Start pipe
+ try (final Connection connection = senderEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ final Statement statement = connection.createStatement()) {
+ statement.execute("start pipe a2b");
+ } catch (SQLException e) {
+ fail(e.getMessage());
+ }
+
+ // Alter pipe (modify)
+ try (final Connection connection = senderEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ final Statement statement = connection.createStatement()) {
+ statement.execute("alter pipe a2b modify sink ('connector.batch.enable'='true')");
+ } catch (SQLException e) {
+ fail(e.getMessage());
+ }
+
+ // Show pipe
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+ final List<TShowPipeInfo> showPipeResult = client.showPipe(new TShowPipeReq()).pipeInfoList;
+ Assert.assertEquals(1, showPipeResult.size());
+ // Check status
+ Assert.assertEquals("RUNNING", showPipeResult.get(0).state);
+ // Check configurations
+ Assert.assertTrue(showPipeResult.get(0).pipeExtractor.contains("source=iotdb-source"));
+ Assert.assertTrue(showPipeResult.get(0).pipeExtractor.contains("database-name=test"));
+ Assert.assertTrue(showPipeResult.get(0).pipeExtractor.contains("table-name=test"));
+ Assert.assertTrue(showPipeResult.get(0).pipeConnector.contains("batch.enable=true"));
+ Assert.assertTrue(
+ showPipeResult
+ .get(0)
+ .pipeConnector
+ .contains(String.format("node-urls=%s", receiverDataNode.getIpAndPortString())));
+ // Check creation time and record last creation time
+ Assert.assertTrue(showPipeResult.get(0).creationTime > lastCreationTime);
+ lastCreationTime = showPipeResult.get(0).creationTime;
+ // Check exception message
+ Assert.assertEquals("", showPipeResult.get(0).exceptionMessage);
+ }
+
+ // Alter pipe (modify empty)
+ try (final Connection connection = senderEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ final Statement statement = connection.createStatement()) {
+ statement.execute("alter pipe a2b modify source ()");
+ } catch (SQLException e) {
+ fail(e.getMessage());
+ }
+
+ // show pipe
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+ final List<TShowPipeInfo> showPipeResult = client.showPipe(new TShowPipeReq()).pipeInfoList;
+ Assert.assertEquals(1, showPipeResult.size());
+ // check status
+ Assert.assertEquals("RUNNING", showPipeResult.get(0).state);
+ // check configurations
+ Assert.assertTrue(showPipeResult.get(0).pipeExtractor.contains("source=iotdb-source"));
+ Assert.assertTrue(showPipeResult.get(0).pipeExtractor.contains("database-name=test"));
+ Assert.assertTrue(showPipeResult.get(0).pipeExtractor.contains("table-name=test"));
+ Assert.assertTrue(showPipeResult.get(0).pipeConnector.contains("batch.enable=true"));
+ Assert.assertTrue(
+ showPipeResult
+ .get(0)
+ .pipeConnector
+ .contains(String.format("node-urls=%s", receiverDataNode.getIpAndPortString())));
+ // Check creation time and record last creation time
+ Assert.assertTrue(showPipeResult.get(0).creationTime > lastCreationTime);
+ lastCreationTime = showPipeResult.get(0).creationTime;
+ // Check exception message
+ Assert.assertEquals("", showPipeResult.get(0).exceptionMessage);
+ }
+
+ // Alter pipe (replace empty)
+ try (final Connection connection = senderEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ final Statement statement = connection.createStatement()) {
+ statement.execute("alter pipe a2b replace source ()");
+ } catch (SQLException e) {
+ fail(e.getMessage());
+ }
+
+ // show pipe
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+ final List<TShowPipeInfo> showPipeResult = client.showPipe(new TShowPipeReq()).pipeInfoList;
+ Assert.assertEquals(1, showPipeResult.size());
+ // check status
+ Assert.assertEquals("RUNNING", showPipeResult.get(0).state);
+ // check configurations
+ Assert.assertFalse(showPipeResult.get(0).pipeExtractor.contains("source=iotdb-source"));
+ Assert.assertFalse(showPipeResult.get(0).pipeExtractor.contains("database-name=test"));
+ Assert.assertFalse(showPipeResult.get(0).pipeExtractor.contains("table-name=test"));
+ Assert.assertTrue(showPipeResult.get(0).pipeConnector.contains("batch.enable=true"));
+ Assert.assertTrue(
+ showPipeResult
+ .get(0)
+ .pipeConnector
+ .contains(String.format("node-urls=%s", receiverDataNode.getIpAndPortString())));
+ // Check creation time and record last creation time
+ Assert.assertTrue(showPipeResult.get(0).creationTime > lastCreationTime);
+ lastCreationTime = showPipeResult.get(0).creationTime;
+ // Check exception message
+ Assert.assertEquals("", showPipeResult.get(0).exceptionMessage);
+ }
+
+ // Alter pipe (replace empty)
+ try (final Connection connection = senderEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ final Statement statement = connection.createStatement()) {
+ statement.execute("alter pipe a2b replace processor ()");
+ } catch (SQLException e) {
+ fail(e.getMessage());
+ }
+
+ // show pipe
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+ final List<TShowPipeInfo> showPipeResult = client.showPipe(new TShowPipeReq()).pipeInfoList;
+ Assert.assertEquals(1, showPipeResult.size());
+ // check status
+ Assert.assertEquals("RUNNING", showPipeResult.get(0).state);
+ // check configurations
+ Assert.assertTrue(showPipeResult.get(0).pipeConnector.contains("batch.enable=true"));
+ Assert.assertFalse(
+ showPipeResult.get(0).pipeProcessor.contains("processor=do-nothing-processor"));
+ Assert.assertTrue(
+ showPipeResult
+ .get(0)
+ .pipeConnector
+ .contains(String.format("node-urls=%s", receiverDataNode.getIpAndPortString())));
+ // Check creation time and record last creation time
+ Assert.assertTrue(showPipeResult.get(0).creationTime > lastCreationTime);
+ lastCreationTime = showPipeResult.get(0).creationTime;
+ // Check exception message
+ Assert.assertEquals("", showPipeResult.get(0).exceptionMessage);
+ }
+
+ // Alter pipe (modify empty)
+ try (final Connection connection = senderEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ final Statement statement = connection.createStatement()) {
+ statement.execute("alter pipe a2b modify sink ()");
+ } catch (SQLException e) {
+ fail(e.getMessage());
+ }
+
+ // Show pipe
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+ final List<TShowPipeInfo> showPipeResult = client.showPipe(new TShowPipeReq()).pipeInfoList;
+ Assert.assertEquals(1, showPipeResult.size());
+ // Check status
+ Assert.assertEquals("RUNNING", showPipeResult.get(0).state);
+ // Check configurations
+ Assert.assertTrue(showPipeResult.get(0).pipeConnector.contains("batch.enable=true"));
+ Assert.assertTrue(
+ showPipeResult
+ .get(0)
+ .pipeConnector
+ .contains(String.format("node-urls=%s", receiverDataNode.getIpAndPortString())));
+ // Check creation time and record last creation time
+ Assert.assertTrue(showPipeResult.get(0).creationTime > lastCreationTime);
+ // Check exception message
+ Assert.assertEquals("", showPipeResult.get(0).exceptionMessage);
+ }
+ }
+
+ @Test
+ public void testAlterPipeFailure() {
+ final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+
+ // alter non-existed pipe
+ String sql =
+ String.format(
+ "alter pipe a2b modify sink ('node-urls'='%s', 'batch.enable'='true')",
+ receiverDataNode.getIpAndPortString());
+ try (final Connection connection = senderEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ final Statement statement = connection.createStatement()) {
+ statement.execute(sql);
+ fail();
+ } catch (SQLException ignore) {
+ }
+
+ // Create pipe
+ sql =
+ String.format(
+ "create pipe a2b with source ('source'='iotdb-source', 'database-name'='test', 'table-name'='test1', 'mode.streaming'='true') with sink ('node-urls'='%s', 'batch.enable'='false')",
+ receiverDataNode.getIpAndPortString());
+ try (final Connection connection = senderEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ final Statement statement = connection.createStatement()) {
+ statement.execute(sql);
+ } catch (SQLException e) {
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testAlterPipeSourceAndSink() {
+ final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+ boolean insertResult = true;
+
+ TableModelUtils.createDataBaseAndTable(senderEnv, "test", "test");
+ TableModelUtils.createDataBaseAndTable(senderEnv, "test1", "test1");
+ // Create pipe
+ final String sql =
+ String.format(
+ "create pipe a2b with source ('source'='iotdb-source', 'database-name'='test', 'table-name'='test', 'mode.streaming'='true') with processor ('processor'='do-nothing-processor') with sink ('node-urls'='%s', 'batch.enable'='false')",
+ receiverDataNode.getIpAndPortString());
+ try (final Connection connection = senderEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ final Statement statement = connection.createStatement()) {
+ statement.execute(sql);
+ } catch (final SQLException e) {
+ fail(e.getMessage());
+ }
+
+ insertResult = TableModelUtils.insertData("test", "test", 0, 100, senderEnv);
+ insertResult = insertResult && TableModelUtils.insertData("test1", "test1", 0, 100, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+
+ // Check data on receiver
+ TableModelUtils.assertData("test", "test", 0, 100, receiverEnv);
+
+ // Alter pipe (modify 'source.path', 'source.inclusion' and
+ // 'processor.tumbling-time.interval-seconds')
+ try (final Connection connection = senderEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ final Statement statement = connection.createStatement()) {
+ statement.execute(
+ "alter pipe a2b modify source('source' = 'iotdb-source','database-name'='test1', 'table-name'='test1', 'mode.streaming'='true', 'source.inclusion'='data.insert') modify sink ('batch.enable'='true')");
+ } catch (final SQLException e) {
+ fail(e.getMessage());
+ }
+ insertResult = TableModelUtils.insertData("test", "test", 100, 200, senderEnv);
+ insertResult =
+ insertResult && TableModelUtils.insertData("test1", "test1", 100, 200, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ TestUtils.assertDataEventuallyOnEnv(
+ receiverEnv,
+ TableModelUtils.getQuerySql("test"),
+ TableModelUtils.generateHeaderResults(),
+ TableModelUtils.generateExpectedResults(0, 100),
+ "test");
+ TestUtils.assertDataEventuallyOnEnv(
+ receiverEnv,
+ TableModelUtils.getQuerySql("test1"),
+ TableModelUtils.generateHeaderResults(),
+ TableModelUtils.generateExpectedResults(0, 200),
+ "test1");
+ }
+}
diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/tablemodel/IoTDBPipeAutoConflictIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/tablemodel/IoTDBPipeAutoConflictIT.java
new file mode 100644
index 00000000000..3eccb7beafd
--- /dev/null
+++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/tablemodel/IoTDBPipeAutoConflictIT.java
@@ -0,0 +1,267 @@
+/*
+ * 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.pipe.it.tablemodel;
+
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.client.sync.SyncConfigNodeIServiceClient;
+import org.apache.iotdb.confignode.rpc.thrift.TCreatePipeReq;
+import org.apache.iotdb.consensus.ConsensusFactory;
+import org.apache.iotdb.db.it.utils.TestUtils;
+import org.apache.iotdb.it.env.MultiEnvFactory;
+import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper;
+import org.apache.iotdb.it.framework.IoTDBTestRunner;
+import org.apache.iotdb.itbase.category.MultiClusterIT2TableModel;
+import org.apache.iotdb.itbase.env.BaseEnv;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.junit.Assert.fail;
+
+@RunWith(IoTDBTestRunner.class)
+@Category({MultiClusterIT2TableModel.class})
+public class IoTDBPipeAutoConflictIT extends AbstractPipeTableModelTestIT {
+ @Before
+ public void setUp() {
+ MultiEnvFactory.createEnv(2);
+ senderEnv = MultiEnvFactory.getEnv(0);
+ receiverEnv = MultiEnvFactory.getEnv(1);
+
+ // TODO: delete ratis configurations
+ senderEnv
+ .getConfig()
+ .getCommonConfig()
+ .setAutoCreateSchemaEnabled(true)
+ .setConfigNodeConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS)
+ .setSchemaRegionConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS)
+ .setDataRegionConsensusProtocolClass(ConsensusFactory.IOT_CONSENSUS);
+ receiverEnv
+ .getConfig()
+ .getCommonConfig()
+ .setAutoCreateSchemaEnabled(true)
+ .setConfigNodeConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS)
+ .setSchemaRegionConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS)
+ .setDataRegionConsensusProtocolClass(ConsensusFactory.IOT_CONSENSUS);
+
+ // 10 min, assert that the operations will not time out
+ senderEnv.getConfig().getCommonConfig().setCnConnectionTimeoutMs(600000);
+ receiverEnv.getConfig().getCommonConfig().setCnConnectionTimeoutMs(600000);
+
+ senderEnv.initClusterEnvironment();
+ receiverEnv.initClusterEnvironment();
+ }
+
+ @Test
+ public void testDoubleLivingAutoConflict() throws Exception {
+ // Double living is two clusters each with a pipe connecting to the other.
+ final DataNodeWrapper senderDataNode = senderEnv.getDataNodeWrapper(0);
+ final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+
+ createDataBaseAndTable(senderEnv);
+ createDataBaseAndTable(receiverEnv);
+
+ final String senderIp = senderDataNode.getIp();
+ final int senderPort = senderDataNode.getPort();
+ final String receiverIp = receiverDataNode.getIp();
+ final int receiverPort = receiverDataNode.getPort();
+
+ insertData("test", "test1", 0, 100, senderEnv);
+
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+ final Map<String, String> extractorAttributes = new HashMap<>();
+ final Map<String, String> processorAttributes = new HashMap<>();
+ final Map<String, String> connectorAttributes = new HashMap<>();
+
+ extractorAttributes.put("source.inclusion", "data.insert");
+ extractorAttributes.put("source.inclusion.exclusion", "");
+ extractorAttributes.put("capture.table", "true");
+ // Add this property to avoid making self cycle.
+ extractorAttributes.put("source.forwarding-pipe-requests", "false");
+
+ connectorAttributes.put("sink", "iotdb-thrift-sink");
+ connectorAttributes.put("sink.batch.enable", "false");
+ connectorAttributes.put("sink.ip", receiverIp);
+ connectorAttributes.put("sink.port", Integer.toString(receiverPort));
+
+ final TSStatus status =
+ client.createPipe(
+ new TCreatePipeReq("p1", connectorAttributes)
+ .setExtractorAttributes(extractorAttributes)
+ .setProcessorAttributes(processorAttributes));
+
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("p1").getCode());
+ }
+
+ insertData("test", "test1", 100, 200, senderEnv);
+ insertData("test", "test1", 200, 300, receiverEnv);
+
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) receiverEnv.getLeaderConfigNodeConnection()) {
+ final Map<String, String> extractorAttributes = new HashMap<>();
+ final Map<String, String> processorAttributes = new HashMap<>();
+ final Map<String, String> connectorAttributes = new HashMap<>();
+
+ extractorAttributes.put("source.inclusion", "data.insert");
+ extractorAttributes.put("source.inclusion.exclusion", "");
+ extractorAttributes.put("capture.table", "true");
+ // Add this property to avoid to make self cycle.
+ extractorAttributes.put("source.forwarding-pipe-requests", "false");
+
+ connectorAttributes.put("connector", "iotdb-thrift-connector");
+ connectorAttributes.put("connector.batch.enable", "false");
+ connectorAttributes.put("connector.ip", senderIp);
+ connectorAttributes.put("connector.port", Integer.toString(senderPort));
+
+ final TSStatus status =
+ client.createPipe(
+ new TCreatePipeReq("p1", connectorAttributes)
+ .setExtractorAttributes(extractorAttributes)
+ .setProcessorAttributes(processorAttributes));
+
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("p1").getCode());
+ }
+ insertData("test", "test1", 300, 400, receiverEnv);
+
+ TableModelUtils.assertData("test", "test1", 0, 400, receiverEnv);
+
+ try {
+ TestUtils.restartCluster(senderEnv);
+ TestUtils.restartCluster(receiverEnv);
+ } catch (final Throwable e) {
+ fail(e.getMessage());
+ }
+
+ insertData("test", "test1", 400, 500, senderEnv);
+ insertData("test", "test1", 500, 600, receiverEnv);
+
+ TableModelUtils.assertData("test", "test1", 0, 600, receiverEnv);
+ }
+
+ @Test
+ public void testDoubleLivingAutoConflictTemplate() throws Exception {
+ final DataNodeWrapper senderDataNode = senderEnv.getDataNodeWrapper(0);
+ final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+
+ final String senderIp = senderDataNode.getIp();
+ final int senderPort = senderDataNode.getPort();
+ final String receiverIp = receiverDataNode.getIp();
+ final int receiverPort = receiverDataNode.getPort();
+
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+
+ createDataBaseAndTable(senderEnv);
+ createDataBaseAndTable(receiverEnv);
+ insertData("test", "test", 0, 100, senderEnv);
+ final Map<String, String> extractorAttributes = new HashMap<>();
+ final Map<String, String> processorAttributes = new HashMap<>();
+ final Map<String, String> connectorAttributes = new HashMap<>();
+
+ extractorAttributes.put("source.inclusion", "data.insert");
+ extractorAttributes.put("source.inclusion.exclusion", "");
+ extractorAttributes.put("table-name", "test.*");
+ extractorAttributes.put("capture.table", "true");
+ extractorAttributes.put("source.forwarding-pipe-requests", "false");
+
+ connectorAttributes.put("connector", "iotdb-thrift-connector");
+ connectorAttributes.put("connector.batch.enable", "false");
+ connectorAttributes.put("connector.ip", receiverIp);
+ connectorAttributes.put("connector.port", Integer.toString(receiverPort));
+
+ final TSStatus status =
+ client.createPipe(
+ new TCreatePipeReq("p1", connectorAttributes)
+ .setExtractorAttributes(extractorAttributes)
+ .setProcessorAttributes(processorAttributes));
+
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("p1").getCode());
+ }
+
+ TableModelUtils.insertData("test", "test", 100, 200, senderEnv);
+ TableModelUtils.insertData("test", "test1", 200, 300, receiverEnv);
+
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) receiverEnv.getLeaderConfigNodeConnection()) {
+ final Map<String, String> extractorAttributes = new HashMap<>();
+ final Map<String, String> processorAttributes = new HashMap<>();
+ final Map<String, String> connectorAttributes = new HashMap<>();
+
+ extractorAttributes.put("source.inclusion", "data.insert");
+ extractorAttributes.put("source.inclusion.exclusion", "");
+ extractorAttributes.put("table-name", "test.*");
+ extractorAttributes.put("capture.table", "true");
+ extractorAttributes.put("source.forwarding-pipe-requests", "false");
+
+ connectorAttributes.put("connector", "iotdb-thrift-connector");
+ connectorAttributes.put("connector.batch.enable", "true");
+ connectorAttributes.put("connector.ip", senderIp);
+ connectorAttributes.put("connector.port", Integer.toString(senderPort));
+
+ final TSStatus status =
+ client.createPipe(
+ new TCreatePipeReq("p1", connectorAttributes)
+ .setExtractorAttributes(extractorAttributes)
+ .setProcessorAttributes(processorAttributes));
+
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("p1").getCode());
+ }
+ insertData("test", "test1", 300, 400, senderEnv);
+
+ TestUtils.assertDataEventuallyOnEnv(
+ receiverEnv,
+ TableModelUtils.getQuerySql("test"),
+ TableModelUtils.generateHeaderResults(),
+ TableModelUtils.generateExpectedResults(0, 200),
+ "test");
+
+ TestUtils.assertDataEventuallyOnEnv(
+ senderEnv,
+ TableModelUtils.getQuerySql("test1"),
+ TableModelUtils.generateHeaderResults(),
+ TableModelUtils.generateExpectedResults(200, 400),
+ "test");
+ }
+
+ private void createDataBaseAndTable(BaseEnv baseEnv) {
+ TableModelUtils.createDataBaseAndTable(baseEnv, "test", "test");
+ TableModelUtils.createDataBaseAndTable(baseEnv, "test1", "test");
+ }
+
+ private void insertData(
+ String dataBaseName, String tableName, int start, int end, BaseEnv baseEnv) {
+ TableModelUtils.insertData(dataBaseName, tableName, start, end, baseEnv);
+ }
+}
diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/tablemodel/IoTDBPipeAutoDropIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/tablemodel/IoTDBPipeAutoDropIT.java
new file mode 100644
index 00000000000..ef4f645b27f
--- /dev/null
+++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/tablemodel/IoTDBPipeAutoDropIT.java
@@ -0,0 +1,152 @@
+/*
+ * 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.pipe.it.tablemodel;
+
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.client.sync.SyncConfigNodeIServiceClient;
+import org.apache.iotdb.confignode.rpc.thrift.TCreatePipeReq;
+import org.apache.iotdb.db.it.utils.TestUtils;
+import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper;
+import org.apache.iotdb.it.framework.IoTDBTestRunner;
+import org.apache.iotdb.itbase.category.MultiClusterIT2TableModel;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+@RunWith(IoTDBTestRunner.class)
+@Category({MultiClusterIT2TableModel.class})
+public class IoTDBPipeAutoDropIT extends AbstractPipeTableModelTestIT {
+
+ @Test
+ public void testAutoDropInHistoricalTransfer() throws Exception {
+ final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+ boolean insertResult = true;
+
+ final String receiverIp = receiverDataNode.getIp();
+ final int receiverPort = receiverDataNode.getPort();
+
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+
+ TableModelUtils.createDataBaseAndTable(senderEnv, "test", "test");
+ insertResult = TableModelUtils.insertData("test", "test", 0, 100, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+
+ final Map<String, String> extractorAttributes = new HashMap<>();
+ final Map<String, String> processorAttributes = new HashMap<>();
+ final Map<String, String> connectorAttributes = new HashMap<>();
+
+ extractorAttributes.put("mode.snapshot", "true");
+ extractorAttributes.put("capture.table", "true");
+
+ connectorAttributes.put("connector", "iotdb-thrift-connector");
+ connectorAttributes.put("connector.batch.enable", "false");
+ connectorAttributes.put("connector.ip", receiverIp);
+ connectorAttributes.put("connector.port", Integer.toString(receiverPort));
+
+ final TSStatus status =
+ client.createPipe(
+ new TCreatePipeReq("p1", connectorAttributes)
+ .setExtractorAttributes(extractorAttributes)
+ .setProcessorAttributes(processorAttributes));
+
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("p1").getCode());
+
+ TestUtils.assertDataEventuallyOnEnv(
+ receiverEnv,
+ TableModelUtils.getQueryCountSql("test"),
+ "_col0,",
+ Collections.singleton("100,"),
+ "test");
+
+ TestUtils.assertDataEventuallyOnEnv(
+ senderEnv,
+ "show pipes",
+ "ID,CreationTime,State,PipeSource,PipeProcessor,PipeSink,ExceptionMessage,RemainingEventCount,EstimatedRemainingSeconds,",
+ Collections.emptySet());
+ }
+ }
+
+ @Test
+ public void testAutoDropInHistoricalTransferWithTimeRange() throws Exception {
+ final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+ boolean insertResult = true;
+
+ final String receiverIp = receiverDataNode.getIp();
+ final int receiverPort = receiverDataNode.getPort();
+
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+
+ TableModelUtils.createDataBaseAndTable(senderEnv, "test", "test");
+ insertResult = TableModelUtils.insertData("test", "test", 0, 100, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ final Map<String, String> extractorAttributes = new HashMap<>();
+ final Map<String, String> processorAttributes = new HashMap<>();
+ final Map<String, String> connectorAttributes = new HashMap<>();
+
+ extractorAttributes.put("mode.snapshot", "true");
+ extractorAttributes.put("capture.table", "true");
+ extractorAttributes.put("start-time", "0");
+ extractorAttributes.put("end-time", "49");
+
+ connectorAttributes.put("connector", "iotdb-thrift-connector");
+ connectorAttributes.put("connector.batch.enable", "false");
+ connectorAttributes.put("connector.ip", receiverIp);
+ connectorAttributes.put("connector.port", Integer.toString(receiverPort));
+
+ final TSStatus status =
+ client.createPipe(
+ new TCreatePipeReq("p1", connectorAttributes)
+ .setExtractorAttributes(extractorAttributes)
+ .setProcessorAttributes(processorAttributes));
+
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("p1").getCode());
+
+ TestUtils.assertDataEventuallyOnEnv(
+ receiverEnv,
+ TableModelUtils.getQueryCountSql("test"),
+ "_col0,",
+ Collections.singleton("50,"),
+ "test");
+
+ TestUtils.assertDataEventuallyOnEnv(
+ senderEnv,
+ "show pipes",
+ "ID,CreationTime,State,PipeSource,PipeProcessor,PipeSink,ExceptionMessage,RemainingEventCount,EstimatedRemainingSeconds,",
+ Collections.emptySet());
+ }
+ }
+}
diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/tablemodel/IoTDBPipeClusterIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/tablemodel/IoTDBPipeClusterIT.java
new file mode 100644
index 00000000000..8ad449eb149
--- /dev/null
+++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/tablemodel/IoTDBPipeClusterIT.java
@@ -0,0 +1,909 @@
+/*
+ * 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.pipe.it.tablemodel;
+
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.client.exception.ClientManagerException;
+import org.apache.iotdb.commons.client.sync.SyncConfigNodeIServiceClient;
+import org.apache.iotdb.commons.cluster.RegionRoleType;
+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.TShowRegionReq;
+import org.apache.iotdb.confignode.rpc.thrift.TShowRegionResp;
+import org.apache.iotdb.consensus.ConsensusFactory;
+import org.apache.iotdb.db.it.utils.TestUtils;
+import org.apache.iotdb.it.env.MultiEnvFactory;
+import org.apache.iotdb.it.env.cluster.env.AbstractEnv;
+import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper;
+import org.apache.iotdb.it.framework.IoTDBTestRunner;
+import org.apache.iotdb.itbase.category.MultiClusterIT2TableModel;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import org.apache.thrift.TException;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.junit.Assert.fail;
+
+@RunWith(IoTDBTestRunner.class)
+@Category({MultiClusterIT2TableModel.class})
+public class IoTDBPipeClusterIT extends AbstractPipeTableModelTestIT {
+
+ @Override
+ @Before
+ public void setUp() {
+ MultiEnvFactory.createEnv(2);
+ senderEnv = MultiEnvFactory.getEnv(0);
+ receiverEnv = MultiEnvFactory.getEnv(1);
+
+ senderEnv
+ .getConfig()
+ .getCommonConfig()
+ .setAutoCreateSchemaEnabled(true)
+ .setConfigNodeConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS)
+ .setSchemaRegionConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS)
+ .setDataRegionConsensusProtocolClass(ConsensusFactory.IOT_CONSENSUS);
+
+ receiverEnv
+ .getConfig()
+ .getCommonConfig()
+ .setAutoCreateSchemaEnabled(true)
+ .setConfigNodeConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS)
+ .setSchemaRegionConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS)
+ .setDataRegionConsensusProtocolClass(ConsensusFactory.IOT_CONSENSUS);
+
+ // 10 min, assert that the operations will not time out
+ senderEnv.getConfig().getCommonConfig().setCnConnectionTimeoutMs(600000);
+ receiverEnv.getConfig().getCommonConfig().setCnConnectionTimeoutMs(600000);
+
+ senderEnv.initClusterEnvironment(3, 3, 180);
+ receiverEnv.initClusterEnvironment(3, 3, 180);
+ }
+
+ @Test
+ public void testWithAllParametersInStreamingMode() throws Exception {
+ testWithAllParameters("true");
+ }
+
+ @Test
+ public void testWithAllParametersInNotStreamingMode() throws Exception {
+ testWithAllParameters("false");
+ }
+
+ private void testWithAllParameters(final String realtimeMode) throws Exception {
+ final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+ boolean insertResult = true;
+ final String receiverIp = receiverDataNode.getIp();
+ final int receiverPort = receiverDataNode.getPort();
+
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+ TableModelUtils.createDataBaseAndTable(senderEnv, "test", "test");
+ insertResult = TableModelUtils.insertData("test", "test", 0, 100, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+
+ final Map<String, String> extractorAttributes = new HashMap<>();
+ final Map<String, String> processorAttributes = new HashMap<>();
+ final Map<String, String> connectorAttributes = new HashMap<>();
+
+ extractorAttributes.put("extractor", "iotdb-extractor");
+ extractorAttributes.put("capture.table", "true");
+ extractorAttributes.put("database-name", "test");
+ extractorAttributes.put("table-name", "test");
+ extractorAttributes.put("start-time", "0");
+ extractorAttributes.put("end-time", "199");
+ extractorAttributes.put("mode.streaming", realtimeMode);
+
+ processorAttributes.put("processor", "do-nothing-processor");
+
+ connectorAttributes.put("connector", "iotdb-thrift-connector");
+ connectorAttributes.put("connector.batch.enable", "false");
+ connectorAttributes.put("connector.ip", receiverIp);
+ connectorAttributes.put("connector.port", Integer.toString(receiverPort));
+ connectorAttributes.put("connector.user", "root");
+ connectorAttributes.put("connector.password", "root");
+
+ final TSStatus status =
+ client.createPipe(
+ new TCreatePipeReq("p1", connectorAttributes)
+ .setExtractorAttributes(extractorAttributes)
+ .setProcessorAttributes(processorAttributes));
+
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("p1").getCode());
+
+ TestUtils.assertDataEventuallyOnEnv(
+ receiverEnv,
+ TableModelUtils.getQuerySql("test"),
+ TableModelUtils.generateHeaderResults(),
+ TableModelUtils.generateExpectedResults(0, 100),
+ "test");
+
+ insertResult = TableModelUtils.insertData("test", "test", 100, 300, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ TestUtils.assertDataEventuallyOnEnv(
+ receiverEnv,
+ TableModelUtils.getQuerySql("test"),
+ TableModelUtils.generateHeaderResults(),
+ TableModelUtils.generateExpectedResults(0, 200),
+ "test");
+ }
+ }
+
+ @Test
+ public void testPipeAfterDataRegionLeaderStop() throws Exception {
+ final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+
+ final String receiverIp = receiverDataNode.getIp();
+ final int receiverPort = receiverDataNode.getPort();
+
+ boolean insertResult = true;
+
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+ final Map<String, String> extractorAttributes = new HashMap<>();
+ final Map<String, String> processorAttributes = new HashMap<>();
+ final Map<String, String> connectorAttributes = new HashMap<>();
+ TableModelUtils.createDataBaseAndTable(senderEnv, "test", "test");
+ TableModelUtils.createDataBaseAndTable(senderEnv, "test1", "test1");
+ insertResult = TableModelUtils.insertData("test", "test", 0, 100, senderEnv);
+ insertResult =
+ insertResult && TableModelUtils.insertData("test1", "test1", 0, 100, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+
+ extractorAttributes.put("extractor", "iotdb-extractor");
+ extractorAttributes.put("database-name", "test");
+ extractorAttributes.put("capture.table", "true");
+ extractorAttributes.put("table-name", "test");
+ extractorAttributes.put("start-time", "0");
+ extractorAttributes.put("end-time", "300");
+
+ connectorAttributes.put("connector", "iotdb-thrift-connector");
+ connectorAttributes.put("connector.batch.enable", "false");
+ connectorAttributes.put("connector.ip", receiverIp);
+ connectorAttributes.put("connector.port", Integer.toString(receiverPort));
+
+ final TSStatus status =
+ client.createPipe(
+ new TCreatePipeReq("p1", connectorAttributes)
+ .setExtractorAttributes(extractorAttributes)
+ .setProcessorAttributes(processorAttributes));
+
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("p1").getCode());
+
+ insertResult = TableModelUtils.insertData("test", "test", 100, 200, senderEnv);
+ insertResult =
+ insertResult && TableModelUtils.insertData("test1", "test1", 100, 200, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+
+ final AtomicInteger leaderPort = new AtomicInteger(-1);
+ final TShowRegionResp showRegionResp = client.showRegion(new TShowRegionReq());
+ showRegionResp
+ .getRegionInfoList()
+ .forEach(
+ regionInfo -> {
+ if (RegionRoleType.Leader.getRoleType().equals(regionInfo.getRoleType())) {
+ leaderPort.set(regionInfo.getClientRpcPort());
+ }
+ });
+
+ int leaderIndex = -1;
+ for (int i = 0; i < 3; ++i) {
+ if (senderEnv.getDataNodeWrapper(i).getPort() == leaderPort.get()) {
+ leaderIndex = i;
+ try {
+ senderEnv.shutdownDataNode(i);
+ } catch (final Throwable e) {
+ e.printStackTrace();
+ return;
+ }
+ try {
+ TimeUnit.SECONDS.sleep(1);
+ } catch (final InterruptedException ignored) {
+ }
+ try {
+ senderEnv.startDataNode(i);
+ ((AbstractEnv) senderEnv).checkClusterStatusWithoutUnknown();
+ } catch (final Throwable e) {
+ e.printStackTrace();
+ return;
+ }
+ }
+ }
+ if (leaderIndex == -1) { // ensure the leader is stopped
+ fail();
+ }
+
+ insertResult = TableModelUtils.insertData("test", "test", 200, 300, senderEnv);
+ insertResult =
+ insertResult && TableModelUtils.insertData("test1", "test1", 200, 300, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+
+ TableModelUtils.assertData("test", "test", 0, 300, receiverEnv);
+ }
+
+ try {
+ TestUtils.restartCluster(senderEnv);
+ TestUtils.restartCluster(receiverEnv);
+ } catch (final Throwable e) {
+ e.printStackTrace();
+ return;
+ }
+
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+ // Create a new pipe and write new data
+ final Map<String, String> extractorAttributes = new HashMap<>();
+ final Map<String, String> processorAttributes = new HashMap<>();
+ final Map<String, String> connectorAttributes = new HashMap<>();
+
+ extractorAttributes.put("database-name", "test1");
+ extractorAttributes.put("capture.table", "true");
+ extractorAttributes.put("table-name", "test1");
+ extractorAttributes.put("start-time", "0");
+ extractorAttributes.put("end-time", "300");
+
+ connectorAttributes.put("connector", "iotdb-thrift-connector");
+ connectorAttributes.put("connector.batch.enable", "false");
+ connectorAttributes.put("connector.ip", receiverIp);
+ connectorAttributes.put("connector.port", Integer.toString(receiverPort));
+
+ final TSStatus status =
+ client.createPipe(
+ new TCreatePipeReq("p2", connectorAttributes)
+ .setExtractorAttributes(extractorAttributes)
+ .setProcessorAttributes(processorAttributes));
+
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("p2").getCode());
+
+ insertResult = TableModelUtils.insertData("test", "test", 300, 400, senderEnv);
+ insertResult =
+ insertResult && TableModelUtils.insertData("test1", "test1", 300, 400, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ TableModelUtils.assertData("test", "test", 0, 301, receiverEnv);
+ TableModelUtils.assertData("test1", "test1", 0, 301, receiverEnv);
+ }
+ }
+
+ @Test
+ public void testPipeAfterRegisterNewDataNode() throws Exception {
+ final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+
+ final String receiverIp = receiverDataNode.getIp();
+ final int receiverPort = receiverDataNode.getPort();
+
+ boolean insertResult = true;
+
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+ final Map<String, String> extractorAttributes = new HashMap<>();
+ final Map<String, String> processorAttributes = new HashMap<>();
+ final Map<String, String> connectorAttributes = new HashMap<>();
+
+ TableModelUtils.createDataBaseAndTable(senderEnv, "test", "test");
+ TableModelUtils.createDataBaseAndTable(senderEnv, "test1", "test1");
+ insertResult = TableModelUtils.insertData("test", "test", 0, 100, senderEnv);
+ insertResult =
+ insertResult && TableModelUtils.insertData("test1", "test1", 0, 100, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+
+ extractorAttributes.put("database-name", "test");
+ extractorAttributes.put("capture.table", "true");
+ extractorAttributes.put("table-name", "test");
+
+ connectorAttributes.put("connector", "iotdb-thrift-connector");
+ connectorAttributes.put("connector.batch.enable", "false");
+ connectorAttributes.put("connector.ip", receiverIp);
+ connectorAttributes.put("connector.port", Integer.toString(receiverPort));
+
+ final TSStatus status =
+ client.createPipe(
+ new TCreatePipeReq("p1", connectorAttributes)
+ .setExtractorAttributes(extractorAttributes)
+ .setProcessorAttributes(processorAttributes));
+
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("p1").getCode());
+
+ insertResult = TableModelUtils.insertData("test", "test", 100, 200, senderEnv);
+ insertResult =
+ insertResult && TableModelUtils.insertData("test1", "test1", 100, 200, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ try {
+ senderEnv.registerNewDataNode(true);
+ } catch (final Throwable e) {
+ e.printStackTrace();
+ return;
+ }
+
+ final DataNodeWrapper newDataNode =
+ senderEnv.getDataNodeWrapper(senderEnv.getDataNodeWrapperList().size() - 1);
+ insertResult = TableModelUtils.insertData("test", "test", 200, 300, senderEnv, newDataNode);
+ insertResult =
+ insertResult
+ && TableModelUtils.insertData("test1", "test1", 200, 300, senderEnv, newDataNode);
+ if (!insertResult) {
+ return;
+ }
+
+ TableModelUtils.assertData("test", "test", 0, 300, receiverEnv);
+ }
+
+ try {
+ TestUtils.restartCluster(senderEnv);
+ TestUtils.restartCluster(receiverEnv);
+ } catch (final Throwable e) {
+ e.printStackTrace();
+ return;
+ }
+
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+ // create a new pipe and write new data
+ final Map<String, String> extractorAttributes = new HashMap<>();
+ final Map<String, String> processorAttributes = new HashMap<>();
+ final Map<String, String> connectorAttributes = new HashMap<>();
+
+ extractorAttributes.put("database-name", "test1");
+ extractorAttributes.put("capture.table", "true");
+ extractorAttributes.put("table-name", "test1");
+
+ connectorAttributes.put("connector", "iotdb-thrift-connector");
+ connectorAttributes.put("connector.batch.enable", "false");
+ connectorAttributes.put("connector.ip", receiverIp);
+ connectorAttributes.put("connector.port", Integer.toString(receiverPort));
+
+ final TSStatus status =
+ client.createPipe(
+ new TCreatePipeReq("p2", connectorAttributes)
+ .setExtractorAttributes(extractorAttributes)
+ .setProcessorAttributes(processorAttributes));
+
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("p2").getCode());
+
+ insertResult = TableModelUtils.insertData("test", "test", 300, 400, senderEnv);
+ insertResult =
+ insertResult && TableModelUtils.insertData("test1", "test1", 300, 400, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ TableModelUtils.assertData("test1", "test1", 0, 400, receiverEnv);
+ TableModelUtils.assertData("test", "test", 0, 400, receiverEnv);
+ }
+ }
+
+ @Test
+ public void testCreatePipeWhenRegisteringNewDataNode() throws Exception {
+ final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+
+ final String receiverIp = receiverDataNode.getIp();
+ final int receiverPort = receiverDataNode.getPort();
+
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+
+ final Map<String, String> extractorAttributes = new HashMap<>();
+ final Map<String, String> processorAttributes = new HashMap<>();
+ final Map<String, String> connectorAttributes = new HashMap<>();
+
+ extractorAttributes.put("database-name", "test1");
+ extractorAttributes.put("capture.table", "true");
+ extractorAttributes.put("table-name", "test1");
+
+ connectorAttributes.put("connector", "iotdb-thrift-connector");
+ connectorAttributes.put("connector.batch.enable", "false");
+ connectorAttributes.put("connector.ip", receiverIp);
+ connectorAttributes.put("connector.port", Integer.toString(receiverPort));
+
+ final Thread t =
+ new Thread(
+ () -> {
+ for (int i = 0; i < 30; ++i) {
+ try {
+ client.createPipe(
+ new TCreatePipeReq("p" + i, connectorAttributes)
+ .setExtractorAttributes(extractorAttributes)
+ .setProcessorAttributes(processorAttributes));
+ } catch (final TException e) {
+ // Not sure if the "createPipe" has succeeded
+ e.printStackTrace();
+ return;
+ }
+ try {
+ Thread.sleep(100);
+ } catch (final Exception ignored) {
+ }
+ }
+ });
+ t.start();
+ try {
+ senderEnv.registerNewDataNode(true);
+ } catch (final Throwable e) {
+ e.printStackTrace();
+ return;
+ }
+ t.join();
+ }
+
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+ final List<TShowPipeInfo> showPipeResult = client.showPipe(new TShowPipeReq()).pipeInfoList;
+ Assert.assertEquals(30, showPipeResult.size());
+ }
+ }
+
+ @Test
+ public void testRegisteringNewDataNodeWhenTransferringData() throws Exception {
+ final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+
+ final String receiverIp = receiverDataNode.getIp();
+ final int receiverPort = receiverDataNode.getPort();
+ boolean insertResult = true;
+
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+
+ TableModelUtils.createDataBaseAndTable(senderEnv, "test", "test");
+ insertResult = TableModelUtils.insertData("test", "test", 0, 100, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ final Map<String, String> extractorAttributes = new HashMap<>();
+ final Map<String, String> processorAttributes = new HashMap<>();
+ final Map<String, String> connectorAttributes = new HashMap<>();
+
+ extractorAttributes.put("database-name", "test");
+ extractorAttributes.put("capture.table", "true");
+ extractorAttributes.put("table-name", "test");
+
+ connectorAttributes.put("connector", "iotdb-thrift-connector");
+ connectorAttributes.put("connector.batch.enable", "false");
+ connectorAttributes.put("connector.ip", receiverIp);
+ connectorAttributes.put("connector.port", Integer.toString(receiverPort));
+
+ final TSStatus status =
+ client.createPipe(
+ new TCreatePipeReq("p1", connectorAttributes)
+ .setExtractorAttributes(extractorAttributes)
+ .setProcessorAttributes(processorAttributes));
+
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("p1").getCode());
+
+ final AtomicInteger succeedNum = new AtomicInteger(0);
+ final Thread t =
+ new Thread(
+ () -> {
+ try {
+ for (int i = 100; i < 200; ++i) {
+ if (TableModelUtils.insertDataNotThrowError(
+ "test", "test", i, i + 1, senderEnv)) {
+ succeedNum.incrementAndGet();
+ Thread.sleep(100);
+ }
+ }
+ } catch (final InterruptedException ignored) {
+ }
+ });
+ t.start();
+ try {
+ senderEnv.registerNewDataNode(true);
+ } catch (final Throwable e) {
+ e.printStackTrace();
+ return;
+ }
+ t.join();
+ if (!TestUtils.tryExecuteNonQueryWithRetry(senderEnv, "flush")) {
+ return;
+ }
+
+ TableModelUtils.assertCountData("test", "test", succeedNum.get() + 100, receiverEnv);
+
+ try {
+ senderEnv.shutdownDataNode(senderEnv.getDataNodeWrapperList().size() - 1);
+ senderEnv.getDataNodeWrapperList().remove(senderEnv.getDataNodeWrapperList().size() - 1);
+ } catch (final Throwable e) {
+ e.printStackTrace();
+ }
+ }
+ }
+
+ @Test
+ public void testRegisteringNewDataNodeAfterTransferringData() throws Exception {
+ final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+
+ final String receiverIp = receiverDataNode.getIp();
+ final int receiverPort = receiverDataNode.getPort();
+ boolean insertResult = true;
+
+ TableModelUtils.createDataBaseAndTable(senderEnv, "test", "test");
+ insertResult = TableModelUtils.insertData("test", "test", 0, 100, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+ final Map<String, String> extractorAttributes = new HashMap<>();
+ final Map<String, String> processorAttributes = new HashMap<>();
+ final Map<String, String> connectorAttributes = new HashMap<>();
+
+ extractorAttributes.put("database-name", "test");
+ extractorAttributes.put("capture.table", "true");
+ extractorAttributes.put("table-name", "test");
+
+ connectorAttributes.put("connector", "iotdb-thrift-connector");
+ connectorAttributes.put("connector.batch.enable", "false");
+ connectorAttributes.put("connector.ip", receiverIp);
+ connectorAttributes.put("connector.port", Integer.toString(receiverPort));
+
+ final TSStatus status =
+ client.createPipe(
+ new TCreatePipeReq("p1", connectorAttributes)
+ .setExtractorAttributes(extractorAttributes)
+ .setProcessorAttributes(processorAttributes));
+
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("p1").getCode());
+
+ int succeedNum = 0;
+ for (int i = 100; i < 200; ++i) {
+ if (TableModelUtils.insertDataNotThrowError("test", "test", i, i + 1, senderEnv)) {
+ succeedNum++;
+ }
+ }
+
+ try {
+ senderEnv.registerNewDataNode(true);
+ } catch (final Throwable e) {
+ e.printStackTrace();
+ return;
+ }
+
+ TableModelUtils.assertCountData("test", "test", succeedNum + 100, receiverEnv);
+
+ try {
+ senderEnv.shutdownDataNode(senderEnv.getDataNodeWrapperList().size() - 1);
+ senderEnv.getDataNodeWrapperList().remove(senderEnv.getDataNodeWrapperList().size() - 1);
+ } catch (final Throwable e) {
+ e.printStackTrace();
+ }
+ }
+ }
+
+ @Test
+ public void testSenderRestartWhenTransferring() throws Exception {
+ final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+
+ final String receiverIp = receiverDataNode.getIp();
+ final int receiverPort = receiverDataNode.getPort();
+ boolean insertResult = true;
+
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+
+ TableModelUtils.createDataBaseAndTable(senderEnv, "test", "test");
+ insertResult = TableModelUtils.insertData("test", "test", 0, 100, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ final Map<String, String> extractorAttributes = new HashMap<>();
+ final Map<String, String> processorAttributes = new HashMap<>();
+ final Map<String, String> connectorAttributes = new HashMap<>();
+
+ extractorAttributes.put("database-name", "test");
+ extractorAttributes.put("capture.table", "true");
+ extractorAttributes.put("table-name", "test");
+
+ connectorAttributes.put("connector", "iotdb-thrift-connector");
+ connectorAttributes.put("connector.batch.enable", "false");
+ connectorAttributes.put("connector.ip", receiverIp);
+ connectorAttributes.put("connector.port", Integer.toString(receiverPort));
+
+ final TSStatus status =
+ client.createPipe(
+ new TCreatePipeReq("p1", connectorAttributes)
+ .setExtractorAttributes(extractorAttributes)
+ .setProcessorAttributes(processorAttributes));
+
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("p1").getCode());
+ }
+
+ int succeedNum = 0;
+ for (int i = 100; i < 200; ++i) {
+ if (TableModelUtils.insertDataNotThrowError("test", "test", i, i + 1, senderEnv)) {
+ succeedNum++;
+ }
+ }
+ if (!TestUtils.tryExecuteNonQueryWithRetry(senderEnv, "flush")) {
+ return;
+ }
+
+ try {
+ TestUtils.restartCluster(senderEnv);
+ } catch (final Throwable e) {
+ e.printStackTrace();
+ return;
+ }
+
+ TableModelUtils.assertCountData("test", "test", succeedNum + 100, receiverEnv);
+ }
+
+ @Test
+ public void testConcurrentlyCreatePipeOfSameName() throws Exception {
+ final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+
+ final String receiverIp = receiverDataNode.getIp();
+ final int receiverPort = receiverDataNode.getPort();
+
+ final Map<String, String> extractorAttributes = new HashMap<>();
+ final Map<String, String> processorAttributes = new HashMap<>();
+ final Map<String, String> connectorAttributes = new HashMap<>();
+
+ extractorAttributes.put("database-name", "test");
+ extractorAttributes.put("capture.table", "true");
+ extractorAttributes.put("table-name", "test");
+
+ connectorAttributes.put("connector", "iotdb-thrift-connector");
+ connectorAttributes.put("connector.batch.enable", "false");
+ connectorAttributes.put("connector.ip", receiverIp);
+ connectorAttributes.put("connector.port", Integer.toString(receiverPort));
+
+ final AtomicInteger successCount = new AtomicInteger(0);
+ final List<Thread> threads = new ArrayList<>();
+ for (int i = 0; i < 10; ++i) {
+ final Thread t =
+ new Thread(
+ () -> {
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+ final TSStatus status =
+ client.createPipe(
+ new TCreatePipeReq("p1", connectorAttributes)
+ .setExtractorAttributes(extractorAttributes)
+ .setProcessorAttributes(processorAttributes));
+ if (status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ successCount.incrementAndGet();
+ }
+ } catch (final InterruptedException e) {
+ Thread.currentThread().interrupt();
+ } catch (final TException | ClientManagerException | IOException e) {
+ e.printStackTrace();
+ } catch (final Exception e) {
+ // Fail iff pipe exception occurs
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ });
+ t.start();
+ threads.add(t);
+ }
+
+ for (Thread t : threads) {
+ t.join();
+ }
+ Assert.assertEquals(1, successCount.get());
+
+ successCount.set(0);
+ for (int i = 0; i < 10; ++i) {
+ final Thread t =
+ new Thread(
+ () -> {
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+ final TSStatus status = client.dropPipe("p1");
+ if (status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ successCount.incrementAndGet();
+ }
+ } catch (final InterruptedException e) {
+ Thread.currentThread().interrupt();
+ } catch (final TException | ClientManagerException | IOException e) {
+ e.printStackTrace();
+ } catch (final Exception e) {
+ // Fail iff pipe exception occurs
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ });
+ t.start();
+ threads.add(t);
+ }
+ for (final Thread t : threads) {
+ t.join();
+ }
+
+ // Assert at least 1 drop operation succeeds
+ Assert.assertTrue(successCount.get() >= 1);
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+ final List<TShowPipeInfo> showPipeResult = client.showPipe(new TShowPipeReq()).pipeInfoList;
+ Assert.assertEquals(0, showPipeResult.size());
+ }
+ }
+
+ @Test
+ public void testCreate10PipesWithSameConnector() throws Exception {
+ testCreatePipesWithSameConnector(10);
+ }
+
+ @Test
+ public void testCreate50PipesWithSameConnector() throws Exception {
+ testCreatePipesWithSameConnector(50);
+ }
+
+ @Test
+ public void testCreate100PipesWithSameConnector() throws Exception {
+ testCreatePipesWithSameConnector(100);
+ }
+
+ private void testCreatePipesWithSameConnector(final int pipeCount) throws Exception {
+ final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+
+ final String receiverIp = receiverDataNode.getIp();
+ final int receiverPort = receiverDataNode.getPort();
+
+ final Map<String, String> extractorAttributes = new HashMap<>();
+ final Map<String, String> processorAttributes = new HashMap<>();
+ final Map<String, String> connectorAttributes = new HashMap<>();
+
+ extractorAttributes.put("database-name", "test");
+ extractorAttributes.put("capture.table", "true");
+ extractorAttributes.put("table-name", "test");
+
+ connectorAttributes.put("connector", "iotdb-thrift-connector");
+ connectorAttributes.put("connector.batch.enable", "false");
+ connectorAttributes.put("connector.ip", receiverIp);
+ connectorAttributes.put("connector.port", Integer.toString(receiverPort));
+
+ final AtomicInteger successCount = new AtomicInteger(0);
+ final List<Thread> threads = new ArrayList<>();
+ for (int i = 0; i < pipeCount; ++i) {
+ final int finalI = i;
+ final Thread t =
+ new Thread(
+ () -> {
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+ final TSStatus status =
+ client.createPipe(
+ new TCreatePipeReq("p" + finalI, connectorAttributes)
+ .setExtractorAttributes(extractorAttributes)
+ .setProcessorAttributes(processorAttributes));
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+ successCount.incrementAndGet();
+ } catch (final InterruptedException e) {
+ e.printStackTrace();
+ Thread.currentThread().interrupt();
+ } catch (final TException | ClientManagerException | IOException e) {
+ e.printStackTrace();
+ } catch (final Exception e) {
+ // Fail iff pipe exception occurs
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ });
+ t.start();
+ threads.add(t);
+ }
+ for (final Thread t : threads) {
+ t.join();
+ }
+
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+ List<TShowPipeInfo> showPipeResult = client.showPipe(new TShowPipeReq()).pipeInfoList;
+ Assert.assertEquals(successCount.get(), showPipeResult.size());
+ showPipeResult =
+ client.showPipe(new TShowPipeReq().setPipeName("p1").setWhereClause(true)).pipeInfoList;
+ Assert.assertEquals(successCount.get(), showPipeResult.size());
+ }
+ }
+
+ @Test
+ public void testNegativeTimestamp() throws Exception {
+ final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+
+ final String receiverIp = receiverDataNode.getIp();
+ final int receiverPort = receiverDataNode.getPort();
+ boolean insertResult = true;
+
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+
+ TableModelUtils.createDataBaseAndTable(senderEnv, "test", "test");
+ insertResult = TableModelUtils.insertData("test", "test", -100, 100, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ final Map<String, String> extractorAttributes = new HashMap<>();
+ final Map<String, String> processorAttributes = new HashMap<>();
+ final Map<String, String> connectorAttributes = new HashMap<>();
+
+ extractorAttributes.put("extractor", "iotdb-extractor");
+ extractorAttributes.put("database-name", "test");
+ extractorAttributes.put("capture.table", "true");
+ extractorAttributes.put("table-name", "test");
+
+ processorAttributes.put("processor", "do-nothing-processor");
+
+ connectorAttributes.put("connector", "iotdb-thrift-connector");
+ connectorAttributes.put("connector.ip", receiverIp);
+ connectorAttributes.put("connector.port", Integer.toString(receiverPort));
+
+ final TSStatus status =
+ client.createPipe(
+ new TCreatePipeReq("p1", connectorAttributes)
+ .setExtractorAttributes(extractorAttributes)
+ .setProcessorAttributes(processorAttributes));
+
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("p1").getCode());
+
+ TableModelUtils.assertData("test", "test", -100, 100, receiverEnv);
+
+ insertResult = TableModelUtils.insertData("test", "test", -200, -100, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ TableModelUtils.assertData("test", "test", -200, 100, receiverEnv);
+ }
+ }
+}
diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/tablemodel/IoTDBPipeConnectorCompressionIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/tablemodel/IoTDBPipeConnectorCompressionIT.java
new file mode 100644
index 00000000000..48af0784992
--- /dev/null
+++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/tablemodel/IoTDBPipeConnectorCompressionIT.java
@@ -0,0 +1,326 @@
+/*
+ * 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.pipe.it.tablemodel;
+
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.client.sync.SyncConfigNodeIServiceClient;
+import org.apache.iotdb.confignode.rpc.thrift.TCreatePipeReq;
+import org.apache.iotdb.confignode.rpc.thrift.TShowPipeInfo;
+import org.apache.iotdb.confignode.rpc.thrift.TShowPipeReq;
+import org.apache.iotdb.consensus.ConsensusFactory;
+import org.apache.iotdb.db.it.utils.TestUtils;
+import org.apache.iotdb.it.env.MultiEnvFactory;
+import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper;
+import org.apache.iotdb.it.framework.IoTDBTestRunner;
+import org.apache.iotdb.itbase.category.MultiClusterIT2TableModel;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import java.sql.Connection;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.Assert.fail;
+
+@RunWith(IoTDBTestRunner.class)
+@Category({MultiClusterIT2TableModel.class})
+public class IoTDBPipeConnectorCompressionIT extends AbstractPipeTableModelTestIT {
+
+ @Override
+ @Before
+ public void setUp() {
+ // Override to enable air-gap
+ MultiEnvFactory.createEnv(2);
+ senderEnv = MultiEnvFactory.getEnv(0);
+ receiverEnv = MultiEnvFactory.getEnv(1);
+
+ senderEnv
+ .getConfig()
+ .getCommonConfig()
+ .setAutoCreateSchemaEnabled(true)
+ .setConfigNodeConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS)
+ .setSchemaRegionConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS);
+ receiverEnv
+ .getConfig()
+ .getCommonConfig()
+ .setAutoCreateSchemaEnabled(true)
+ .setPipeAirGapReceiverEnabled(true)
+ .setConfigNodeConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS)
+ .setSchemaRegionConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS);
+
+ // 10 min, assert that the operations will not time out
+ senderEnv.getConfig().getCommonConfig().setCnConnectionTimeoutMs(600000);
+ receiverEnv.getConfig().getCommonConfig().setCnConnectionTimeoutMs(600000);
+
+ senderEnv.initClusterEnvironment();
+ receiverEnv.initClusterEnvironment();
+ }
+
+ @Test
+ public void testCompression1() throws Exception {
+ doTest("iotdb-thrift-connector", "stream", true, "snappy");
+ }
+
+ @Test
+ public void testCompression2() throws Exception {
+ doTest("iotdb-thrift-connector", "batch", true, "snappy, lzma2");
+ }
+
+ @Test
+ public void testCompression3() throws Exception {
+ doTest("iotdb-thrift-sync-connector", "stream", false, "snappy, snappy");
+ }
+
+ @Test
+ public void testCompression4() throws Exception {
+ doTest("iotdb-thrift-sync-connector", "batch", true, "gzip, zstd");
+ }
+
+ @Test
+ public void testCompression5() throws Exception {
+ doTest("iotdb-air-gap-connector", "stream", false, "lzma2, lz4");
+ }
+
+ @Test
+ public void testCompression6() throws Exception {
+ doTest("iotdb-air-gap-connector", "batch", true, "lzma2");
+ }
+
+ private void doTest(
+ String connectorType, String realtimeMode, boolean useBatchMode, String compressionTypes)
+ throws Exception {
+ final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+
+ final String receiverIp = receiverDataNode.getIp();
+ final int receiverPort =
+ connectorType.contains("air-gap")
+ ? receiverDataNode.getPipeAirGapReceiverPort()
+ : receiverDataNode.getPort();
+
+ TableModelUtils.createDataBaseAndTable(senderEnv, "test", "test");
+ TableModelUtils.insertData("test", "test", 0, 50, senderEnv, true);
+
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+ if (!TestUtils.tryExecuteNonQueriesWithRetry(
+ senderEnv,
+ Arrays.asList(
+ "insert into root.db.d1(time, s1) values (2010-01-01T10:00:00+08:00, 1)",
+ "insert into root.db.d1(time, s1) values (2010-01-02T10:00:00+08:00, 2)",
+ "flush"))) {
+ return;
+ }
+ final Map<String, String> extractorAttributes = new HashMap<>();
+ final Map<String, String> processorAttributes = new HashMap<>();
+ final Map<String, String> connectorAttributes = new HashMap<>();
+
+ extractorAttributes.put("extractor", "iotdb-extractor");
+ extractorAttributes.put("extractor.realtime.mode", realtimeMode);
+ extractorAttributes.put("capture.table", "true");
+ extractorAttributes.put("capture.tree", "true");
+
+ processorAttributes.put("processor", "do-nothing-processor");
+
+ connectorAttributes.put("connector", connectorType);
+ connectorAttributes.put("connector.batch.enable", useBatchMode ? "true" : "false");
+ connectorAttributes.put("connector.ip", receiverIp);
+ connectorAttributes.put("connector.port", Integer.toString(receiverPort));
+ connectorAttributes.put("connector.user", "root");
+ connectorAttributes.put("connector.password", "root");
+ connectorAttributes.put("connector.compressor", compressionTypes);
+
+ final TSStatus status =
+ client.createPipe(
+ new TCreatePipeReq("p1", connectorAttributes)
+ .setExtractorAttributes(extractorAttributes)
+ .setProcessorAttributes(processorAttributes));
+
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("p1").getCode());
+
+ TestUtils.assertDataEventuallyOnEnv(
+ receiverEnv,
+ "select count(*) from root.**",
+ "count(root.db.d1.s1),",
+ Collections.singleton("2,"));
+
+ if (!TestUtils.tryExecuteNonQueriesWithRetry(
+ senderEnv,
+ Arrays.asList(
+ "insert into root.db.d1(time, s1) values (3, 3)",
+ "insert into root.db.d1(time, s1) values (4, 4)",
+ "insert into root.db.d1(time, s1) values (5, 5)",
+ "insert into root.db.d1(time, s1) values (6, 6)",
+ "insert into root.db.d1(time, s1) values (7, 7)",
+ "insert into root.db.d1(time, s1) values (8, 8)",
+ "flush"))) {
+ return;
+ }
+
+ TableModelUtils.insertData("test", "test", 50, 100, senderEnv, true);
+
+ TestUtils.assertDataEventuallyOnEnv(
+ receiverEnv,
+ "select count(*) from root.**",
+ "count(root.db.d1.s1),",
+ Collections.singleton("8,"));
+
+ TableModelUtils.assertCountData("test", "test", 100, receiverEnv);
+ }
+ }
+
+ @Test
+ public void testZstdCompressorLevel() throws Exception {
+ final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+
+ final String receiverIp = receiverDataNode.getIp();
+ final int receiverPort = receiverDataNode.getPort();
+
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+ if (!TestUtils.tryExecuteNonQueriesWithRetry(
+ senderEnv,
+ Arrays.asList(
+ "insert into root.db.d1(time, s1) values (1, 1)",
+ "insert into root.db.d1(time, s2) values (1, 1)",
+ "insert into root.db.d1(time, s3) values (1, 1)",
+ "insert into root.db.d1(time, s4) values (1, 1)",
+ "insert into root.db.d1(time, s5) values (1, 1)",
+ "flush"))) {
+ return;
+ }
+
+ // Create 5 pipes with different zstd compression levels, p4 and p5 should fail.
+
+ try (final Connection connection = senderEnv.getConnection();
+ final Statement statement = connection.createStatement()) {
+ statement.execute(
+ String.format(
+ "create pipe p1"
+ + " with extractor ('extractor.pattern'='root.db.d1.s1','table-name'='test1','capture.table'='true','capture.tree'='true')"
+ + " with connector ("
+ + "'connector.ip'='%s',"
+ + "'connector.port'='%s',"
+ + "'connector.compressor'='zstd, zstd',"
+ + "'connector.compressor.zstd.level'='3',"
+ + "'connector.rate-limit-bytes-per-second'='2048.0')",
+ receiverIp, receiverPort));
+ } catch (SQLException e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+
+ try (final Connection connection = senderEnv.getConnection();
+ final Statement statement = connection.createStatement()) {
+ statement.execute(
+ String.format(
+ "create pipe p2"
+ + " with extractor ('extractor.pattern'='root.db.d1.s2','table-name'='test2','capture.table'='true','capture.tree'='true')"
+ + " with connector ("
+ + "'connector.ip'='%s',"
+ + "'connector.port'='%s',"
+ + "'connector.compressor'='zstd, zstd',"
+ + "'connector.compressor.zstd.level'='22')",
+ receiverIp, receiverPort));
+ } catch (SQLException e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+
+ try (final Connection connection = senderEnv.getConnection();
+ final Statement statement = connection.createStatement()) {
+ statement.execute(
+ String.format(
+ "create pipe p3"
+ + " with extractor ('extractor.pattern'='root.db.d1.s3','table-name'='test3','capture.table'='true','capture.tree'='true')"
+ + " with connector ("
+ + "'connector.ip'='%s',"
+ + "'connector.port'='%s',"
+ + "'connector.compressor'='zstd, zstd',"
+ + "'connector.compressor.zstd.level'='-131072')",
+ receiverIp, receiverPort));
+ } catch (SQLException e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+
+ try (final Connection connection = senderEnv.getConnection();
+ final Statement statement = connection.createStatement()) {
+ statement.execute(
+ String.format(
+ "create pipe p4"
+ + " with extractor ('extractor.pattern'='root.db.d1.s4','table-name'='test4','capture.table'='true','capture.tree'='true')"
+ + " with connector ("
+ + "'connector.ip'='%s',"
+ + "'connector.port'='%s',"
+ + "'connector.compressor'='zstd, zstd',"
+ + "'connector.compressor.zstd.level'='-131073')",
+ receiverIp, receiverPort));
+ fail();
+ } catch (SQLException e) {
+ // Make sure the error message in IoTDBConnector.java is returned
+ Assert.assertTrue(e.getMessage().contains("Zstd compression level should be in the range"));
+ }
+
+ try (final Connection connection = senderEnv.getConnection();
+ final Statement statement = connection.createStatement()) {
+ statement.execute(
+ String.format(
+ "create pipe p5"
+ + " with extractor ('extractor.pattern'='root.db.d1.s5','table-name'='test5','capture.table'='true','capture.tree'='true')"
+ + " with connector ("
+ + "'connector.ip'='%s',"
+ + "'connector.port'='%s',"
+ + "'connector.compressor'='zstd, zstd',"
+ + "'connector.compressor.zstd.level'='23')",
+ receiverIp, receiverPort));
+ fail();
+ } catch (SQLException e) {
+ // Make sure the error message in IoTDBConnector.java is returned
+ Assert.assertTrue(e.getMessage().contains("Zstd compression level should be in the range"));
+ }
+
+ final List<TShowPipeInfo> showPipeResult = client.showPipe(new TShowPipeReq()).pipeInfoList;
+ Assert.assertEquals(3, showPipeResult.size());
+
+ TableModelUtils.createDataBaseAndTable(senderEnv, "test1", "test");
+ TableModelUtils.insertData("test", "test1", 0, 50, senderEnv, true);
+
+ TableModelUtils.createDataBaseAndTable(senderEnv, "test2", "test");
+ TableModelUtils.insertData("test", "test2", 0, 50, senderEnv, true);
+
+ TableModelUtils.createDataBaseAndTable(senderEnv, "test3", "test");
+ TableModelUtils.insertData("test", "test3", 0, 50, senderEnv, true);
+
+ TableModelUtils.assertCountData("test", "test1", 50, receiverEnv);
+ }
+ }
+}
diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/tablemodel/IoTDBPipeConnectorParallelIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/tablemodel/IoTDBPipeConnectorParallelIT.java
new file mode 100644
index 00000000000..674bf03c2be
--- /dev/null
+++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/tablemodel/IoTDBPipeConnectorParallelIT.java
@@ -0,0 +1,99 @@
+/*
+ * 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.pipe.it.tablemodel;
+
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.client.sync.SyncConfigNodeIServiceClient;
+import org.apache.iotdb.confignode.rpc.thrift.TCreatePipeReq;
+import org.apache.iotdb.db.it.utils.TestUtils;
+import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper;
+import org.apache.iotdb.it.framework.IoTDBTestRunner;
+import org.apache.iotdb.itbase.category.MultiClusterIT2TableModel;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+@RunWith(IoTDBTestRunner.class)
+@Category({MultiClusterIT2TableModel.class})
+public class IoTDBPipeConnectorParallelIT extends AbstractPipeTableModelTestIT {
+ @Test
+ public void testIoTConnectorParallel() throws Exception {
+ final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+
+ final String receiverIp = receiverDataNode.getIp();
+ final int receiverPort = receiverDataNode.getPort();
+ TableModelUtils.createDataBaseAndTable(senderEnv, "test", "test");
+ TableModelUtils.insertData("test", "test", 0, 100, senderEnv);
+ final Set<String> expectedResSet = new HashSet<>();
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+ final Map<String, String> extractorAttributes = new HashMap<>();
+ final Map<String, String> processorAttributes = new HashMap<>();
+ final Map<String, String> connectorAttributes = new HashMap<>();
+
+ extractorAttributes.put("capture.table", "true");
+ connectorAttributes.put("connector", "iotdb-thrift-connector");
+ connectorAttributes.put("connector.batch.enable", "true");
+ connectorAttributes.put("connector.ip", receiverIp);
+ connectorAttributes.put("connector.port", Integer.toString(receiverPort));
+ connectorAttributes.put("connector.parallel.tasks", "3");
+
+ final TSStatus status =
+ client.createPipe(
+ new TCreatePipeReq("testPipe", connectorAttributes)
+ .setExtractorAttributes(extractorAttributes)
+ .setProcessorAttributes(processorAttributes));
+
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("testPipe").getCode());
+
+ if (!TestUtils.tryExecuteNonQueriesWithRetry(
+ senderEnv,
+ Arrays.asList(
+ "insert into root.sg1.d1(time, s1) values (0, 1)",
+ "insert into root.sg1.d1(time, s1) values (1, 2)",
+ "insert into root.sg1.d1(time, s1) values (2, 3)",
+ "insert into root.sg1.d1(time, s1) values (3, 4)",
+ "flush"))) {
+ return;
+ }
+ TableModelUtils.insertData("test", "test", 100, 200, senderEnv);
+ expectedResSet.add("0,1.0,");
+ expectedResSet.add("1,2.0,");
+ expectedResSet.add("2,3.0,");
+ expectedResSet.add("3,4.0,");
+ TestUtils.assertDataEventuallyOnEnv(
+ receiverEnv, "select * from root.**", "Time,root.sg1.d1.s1,", expectedResSet);
+
+ TableModelUtils.assertCountData("test", "test", 200, receiverEnv);
+ }
+ }
+}
diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/tablemodel/IoTDBPipeDataSinkIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/tablemodel/IoTDBPipeDataSinkIT.java
new file mode 100644
index 00000000000..5c936041743
--- /dev/null
+++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/tablemodel/IoTDBPipeDataSinkIT.java
@@ -0,0 +1,219 @@
+/*
+ * 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.pipe.it.tablemodel;
+
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.client.sync.SyncConfigNodeIServiceClient;
+import org.apache.iotdb.confignode.rpc.thrift.TCreatePipeReq;
+import org.apache.iotdb.db.it.utils.TestUtils;
+import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper;
+import org.apache.iotdb.it.framework.IoTDBTestRunner;
+import org.apache.iotdb.itbase.category.MultiClusterIT2TableModel;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import org.junit.Assert;
+import org.junit.Ignore;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+
+@RunWith(IoTDBTestRunner.class)
+@Category({MultiClusterIT2TableModel.class})
+public class IoTDBPipeDataSinkIT extends AbstractPipeTableModelTestIT {
+ @Test
+ public void testThriftConnectorWithRealtimeFirstDisabled() throws Exception {
+ final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+
+ final String receiverIp = receiverDataNode.getIp();
+ final int receiverPort = receiverDataNode.getPort();
+
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+
+ TableModelUtils.createDataBaseAndTable(senderEnv, "test", "test");
+ TableModelUtils.insertData("test", "test", 0, 50, senderEnv, true);
+
+ if (!TestUtils.tryExecuteNonQueriesWithRetry(
+ senderEnv,
+ Arrays.asList("insert into root.vehicle.d0(time, s1) values (0, 1)", "flush"))) {
+ return;
+ }
+
+ final Map<String, String> extractorAttributes = new HashMap<>();
+ final Map<String, String> processorAttributes = new HashMap<>();
+ final Map<String, String> connectorAttributes = new HashMap<>();
+
+ extractorAttributes.put("extractor.realtime.mode", "log");
+ extractorAttributes.put("capture.table", "true");
+ extractorAttributes.put("capture.tree", "true");
+
+ connectorAttributes.put("connector", "iotdb-thrift-connector");
+ connectorAttributes.put("connector.batch.enable", "true");
+ connectorAttributes.put("connector.ip", receiverIp);
+ connectorAttributes.put("connector.port", Integer.toString(receiverPort));
+ connectorAttributes.put("connector.realtime-first", "false");
+
+ final TSStatus status =
+ client.createPipe(
+ new TCreatePipeReq("testPipe", connectorAttributes)
+ .setExtractorAttributes(extractorAttributes)
+ .setProcessorAttributes(processorAttributes));
+
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("testPipe").getCode());
+
+ TableModelUtils.insertData("test", "test", 50, 100, senderEnv, true);
+
+ if (!TestUtils.tryExecuteNonQueriesWithRetry(
+ senderEnv,
+ Arrays.asList("insert into root.vehicle.d0(time, s1) values (1, 1)", "flush"))) {
+ return;
+ }
+
+ TestUtils.assertDataEventuallyOnEnv(
+ receiverEnv,
+ "select * from root.**",
+ "Time,root.vehicle.d0.s1,",
+ Collections.unmodifiableSet(new HashSet<>(Arrays.asList("0,1.0,", "1,1.0,"))));
+
+ TableModelUtils.assertCountData("test", "test", 100, receiverEnv);
+ }
+ }
+
+ @Test
+ public void testSinkTabletFormat() throws Exception {
+ testSinkFormat("tablet");
+ }
+
+ // table model not support
+ @Ignore
+ @Test
+ public void testSinkTsFileFormat() throws Exception {
+ testSinkFormat("tsfile");
+ }
+
+ @Test
+ public void testSinkHybridFormat() throws Exception {
+ testSinkFormat("hybrid");
+ }
+
+ private void testSinkFormat(final String format) throws Exception {
+ final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+
+ final String receiverIp = receiverDataNode.getIp();
+ final int receiverPort = receiverDataNode.getPort();
+
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+
+ TableModelUtils.createDataBaseAndTable(senderEnv, "test", "test");
+ TableModelUtils.insertData("test", "test", 0, 50, senderEnv, true);
+
+ if (!TestUtils.tryExecuteNonQueriesWithRetry(
+ senderEnv,
+ Arrays.asList("insert into root.vehicle.d0(time, s1) values (1, 1)", "flush"))) {
+ return;
+ }
+
+ final Map<String, String> extractorAttributes = new HashMap<>();
+ final Map<String, String> processorAttributes = new HashMap<>();
+ final Map<String, String> connectorAttributes = new HashMap<>();
+
+ extractorAttributes.put("extractor.realtime.mode", "forced-log");
+ extractorAttributes.put("capture.table", "true");
+ extractorAttributes.put("capture.tree", "true");
+
+ connectorAttributes.put("connector", "iotdb-thrift-connector");
+ connectorAttributes.put("connector.batch.enable", "true");
+ connectorAttributes.put("connector.ip", receiverIp);
+ connectorAttributes.put("connector.port", Integer.toString(receiverPort));
+ connectorAttributes.put("connector.format", format);
+ connectorAttributes.put("connector.realtime-first", "false");
+
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
+ client
+ .createPipe(
+ new TCreatePipeReq("testPipe", connectorAttributes)
+ .setExtractorAttributes(extractorAttributes)
+ .setProcessorAttributes(processorAttributes))
+ .getCode());
+
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("testPipe").getCode());
+
+ TableModelUtils.insertData("test", "test", 100, 150, senderEnv, true);
+
+ if (!TestUtils.tryExecuteNonQueriesWithRetry(
+ senderEnv,
+ Arrays.asList("insert into root.vehicle.d0(time, s1) values (2, 1)", "flush"))) {
+ return;
+ }
+
+ TestUtils.assertDataEventuallyOnEnv(
+ receiverEnv,
+ "select * from root.**",
+ "Time,root.vehicle.d0.s1,",
+ Collections.unmodifiableSet(new HashSet<>(Arrays.asList("1,1.0,", "2,1.0,"))));
+
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.dropPipe("testPipe").getCode());
+
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
+ client
+ .createPipe(
+ new TCreatePipeReq("testPipe", connectorAttributes)
+ .setExtractorAttributes(extractorAttributes)
+ .setProcessorAttributes(processorAttributes))
+ .getCode());
+
+ // Do not fail if the failure has nothing to do with pipe
+ // Because the failures will randomly generate due to resource limitation
+ if (!TestUtils.tryExecuteNonQueriesWithRetry(
+ senderEnv,
+ Arrays.asList(
+ "insert into root.vehicle.d0(time, s1) values (4, 1)",
+ "insert into root.vehicle.d0(time, s1) values (3, 1), (0, 1)",
+ "flush"))) {
+ return;
+ }
+
+ TableModelUtils.insertData("test", "test", 150, 200, senderEnv, true);
+
+ TestUtils.assertDataEventuallyOnEnv(
+ receiverEnv,
+ "select * from root.**",
+ "Time,root.vehicle.d0.s1,",
+ Collections.unmodifiableSet(
+ new HashSet<>(Arrays.asList("0,1.0,", "1,1.0,", "2,1.0,", "3,1.0,", "4,1.0,"))));
+
+ TableModelUtils.assertCountData("test", "test", 150, receiverEnv);
+ }
+ }
+}
diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/tablemodel/IoTDBPipeExtractorIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/tablemodel/IoTDBPipeExtractorIT.java
new file mode 100644
index 00000000000..71389c7957f
--- /dev/null
+++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/tablemodel/IoTDBPipeExtractorIT.java
@@ -0,0 +1,921 @@
+/*
+ * 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.pipe.it.tablemodel;
+
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.client.sync.SyncConfigNodeIServiceClient;
+import org.apache.iotdb.confignode.rpc.thrift.TCreatePipeReq;
+import org.apache.iotdb.consensus.ConsensusFactory;
+import org.apache.iotdb.db.it.utils.TestUtils;
+import org.apache.iotdb.it.env.MultiEnvFactory;
+import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper;
+import org.apache.iotdb.it.framework.IoTDBTestRunner;
+import org.apache.iotdb.itbase.category.MultiClusterIT2TableModel;
+import org.apache.iotdb.itbase.env.BaseEnv;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import java.sql.Connection;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.junit.Assert.fail;
+
+@RunWith(IoTDBTestRunner.class)
+@Category({MultiClusterIT2TableModel.class})
+public class IoTDBPipeExtractorIT extends AbstractPipeTableModelTestIT {
+
+ @Before
+ public void setUp() {
+ MultiEnvFactory.createEnv(2);
+ senderEnv = MultiEnvFactory.getEnv(0);
+ receiverEnv = MultiEnvFactory.getEnv(1);
+
+ // TODO: delete ratis configurations
+ senderEnv
+ .getConfig()
+ .getCommonConfig()
+ .setAutoCreateSchemaEnabled(true)
+ .setConfigNodeConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS)
+ .setSchemaRegionConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS)
+ // Disable sender compaction for tsfile determination in loose range test
+ .setEnableSeqSpaceCompaction(false)
+ .setEnableUnseqSpaceCompaction(false)
+ .setEnableCrossSpaceCompaction(false);
+ receiverEnv
+ .getConfig()
+ .getCommonConfig()
+ .setAutoCreateSchemaEnabled(true)
+ .setConfigNodeConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS)
+ .setSchemaRegionConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS);
+
+ // 10 min, assert that the operations will not time out
+ senderEnv.getConfig().getCommonConfig().setCnConnectionTimeoutMs(600000);
+ receiverEnv.getConfig().getCommonConfig().setCnConnectionTimeoutMs(600000);
+
+ senderEnv.initClusterEnvironment();
+ receiverEnv.initClusterEnvironment();
+ }
+
+ @Test
+ public void testMatchingMultipleDatabases() throws Exception {
+ final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+ boolean insertResult = true;
+ final String receiverIp = receiverDataNode.getIp();
+ final int receiverPort = receiverDataNode.getPort();
+
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+ final Map<String, String> extractorAttributes = new HashMap<>();
+ final Map<String, String> processorAttributes = new HashMap<>();
+ final Map<String, String> connectorAttributes = new HashMap<>();
+
+ extractorAttributes.put("extractor.capture.table", "true");
+ extractorAttributes.put("extractor.capture.tree", "true");
+ extractorAttributes.put("extractor.database-name", "test");
+ extractorAttributes.put("extractor.table-name", "test");
+ extractorAttributes.put("extractor.pattern", "root.db1");
+ extractorAttributes.put("extractor.inclusion", "data.insert");
+
+ connectorAttributes.put("connector", "iotdb-thrift-connector");
+ connectorAttributes.put("connector.batch.enable", "false");
+ connectorAttributes.put("connector.ip", receiverIp);
+ connectorAttributes.put("connector.port", Integer.toString(receiverPort));
+
+ TSStatus status =
+ client.createPipe(
+ new TCreatePipeReq("p1", connectorAttributes)
+ .setExtractorAttributes(extractorAttributes)
+ .setProcessorAttributes(processorAttributes));
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("p1").getCode());
+ assertTimeseriesCountOnReceiver(receiverEnv, 0);
+
+ if (!TestUtils.tryExecuteNonQueriesWithRetry(
+ senderEnv,
+ Arrays.asList(
+ "insert into root.db1.d1 (time, at1) values (1, 10)",
+ "insert into root.db2.d1 (time, at1) values (1, 20)",
+ "flush"))) {
+ return;
+ }
+
+ TableModelUtils.createDataBaseAndTable(senderEnv, "test", "test");
+ TableModelUtils.createDataBaseAndTable(senderEnv, "test1", "test");
+ insertResult = TableModelUtils.insertData("test", "test", 0, 100, senderEnv);
+ insertResult =
+ insertResult && TableModelUtils.insertData("test1", "test1", 0, 100, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ extractorAttributes.replace("extractor.pattern", "root.db2");
+ extractorAttributes.replace("extractor.table-name", "test1");
+ status =
+ client.createPipe(
+ new TCreatePipeReq("p2", connectorAttributes)
+ .setExtractorAttributes(extractorAttributes)
+ .setProcessorAttributes(processorAttributes));
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("p2").getCode());
+ assertTimeseriesCountOnReceiver(receiverEnv, 2);
+
+ Thread.sleep(10000);
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.dropPipe("p1").getCode());
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.dropPipe("p2").getCode());
+
+ if (!TestUtils.tryExecuteNonQueriesWithRetry(
+ senderEnv,
+ Arrays.asList(
+ "insert into root.db1.d1 (time, at1) values (2, 11)",
+ "insert into root.db2.d1 (time, at1) values (2, 21)",
+ "flush"))) {
+ return;
+ }
+
+ insertResult = TableModelUtils.insertData("test", "test", 100, 200, senderEnv);
+ insertResult =
+ insertResult && TableModelUtils.insertData("test1", "test1", 100, 200, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ extractorAttributes.remove("extractor.pattern"); // no pattern, will match all databases
+ extractorAttributes.remove("extractor.table-name");
+ extractorAttributes.remove("extractor.database-name");
+ status =
+ client.createPipe(
+ new TCreatePipeReq("p3", connectorAttributes)
+ .setExtractorAttributes(extractorAttributes)
+ .setProcessorAttributes(processorAttributes));
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("p3").getCode());
+
+ TestUtils.assertDataEventuallyOnEnv(
+ receiverEnv,
+ "select count(*) from root.**",
+ "count(root.db1.d1.at1),count(root.db2.d1.at1),",
+ Collections.singleton("2,2,"));
+
+ TableModelUtils.assertCountData("test", "test", 200, receiverEnv);
+ }
+ }
+
+ @Test
+ public void testHistoryAndRealtime() throws Exception {
+ final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+
+ final String receiverIp = receiverDataNode.getIp();
+ final int receiverPort = receiverDataNode.getPort();
+ boolean insertResult = true;
+
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+ if (!TestUtils.tryExecuteNonQueriesWithRetry(
+ senderEnv,
+ Arrays.asList(
+ "insert into root.db.d1 (time, at1) values (1, 10)",
+ "insert into root.db.d2 (time, at1) values (1, 20)",
+ "insert into root.db.d3 (time, at1) values (1, 30)",
+ "insert into root.db.d4 (time, at1) values (1, 40)",
+ "flush"))) {
+ return;
+ }
+
+ TableModelUtils.createDataBaseAndTable(senderEnv, "test1", "test");
+ insertResult = TableModelUtils.insertData("test", "test1", 0, 100, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ TableModelUtils.createDataBaseAndTable(senderEnv, "test2", "test");
+ insertResult = TableModelUtils.insertData("test", "test2", 0, 100, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ TableModelUtils.createDataBaseAndTable(senderEnv, "test3", "test3");
+ insertResult = TableModelUtils.insertData("test", "test3", 0, 100, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ TableModelUtils.createDataBaseAndTable(senderEnv, "test4", "test");
+ insertResult = TableModelUtils.insertData("test", "test4", 0, 100, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ final Map<String, String> extractorAttributes = new HashMap<>();
+ final Map<String, String> processorAttributes = new HashMap<>();
+ final Map<String, String> connectorAttributes = new HashMap<>();
+
+ connectorAttributes.put("connector", "iotdb-thrift-connector");
+ connectorAttributes.put("connector.batch.enable", "false");
+ connectorAttributes.put("connector.ip", receiverIp);
+ connectorAttributes.put("connector.port", Integer.toString(receiverPort));
+
+ extractorAttributes.put("extractor.capture.table", "true");
+ extractorAttributes.put("extractor.capture.tree", "true");
+ extractorAttributes.put("extractor.database", "test");
+ extractorAttributes.put("extractor.table", "test2");
+ extractorAttributes.put("extractor.inclusion", "data.insert");
+ extractorAttributes.put("extractor.pattern", "root.db.d2");
+ extractorAttributes.put("extractor.history.enable", "false");
+ extractorAttributes.put("extractor.realtime.enable", "true");
+ TSStatus status =
+ client.createPipe(
+ new TCreatePipeReq("p2", connectorAttributes)
+ .setExtractorAttributes(extractorAttributes)
+ .setProcessorAttributes(processorAttributes));
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("p2").getCode());
+
+ extractorAttributes.replace("extractor.table-name", "test3");
+ extractorAttributes.replace("extractor.pattern", "root.db.d3");
+ extractorAttributes.replace("extractor.history.enable", "true");
+ extractorAttributes.replace("extractor.realtime.enable", "false");
+ status =
+ client.createPipe(
+ new TCreatePipeReq("p3", connectorAttributes)
+ .setExtractorAttributes(extractorAttributes)
+ .setProcessorAttributes(processorAttributes));
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("p3").getCode());
+
+ extractorAttributes.replace("extractor.table-name", "test4");
+ extractorAttributes.replace("extractor.pattern", "root.db.d4");
+ extractorAttributes.replace("extractor.history.enable", "true");
+ extractorAttributes.replace("extractor.realtime.enable", "true");
+ status =
+ client.createPipe(
+ new TCreatePipeReq("p4", connectorAttributes)
+ .setExtractorAttributes(extractorAttributes)
+ .setProcessorAttributes(processorAttributes));
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("p4").getCode());
+
+ if (!TestUtils.tryExecuteNonQueriesWithRetry(
+ senderEnv,
+ Arrays.asList(
+ "insert into root.db.d1 (time, at1) values (2, 11)",
+ "insert into root.db.d2 (time, at1) values (2, 21)",
+ "insert into root.db.d3 (time, at1) values (2, 31)",
+ "insert into root.db.d4 (time, at1) values (2, 41), (3, 51)"))) {
+ return;
+ }
+
+ insertResult = TableModelUtils.insertData("test", "test2", 100, 200, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ insertResult = TableModelUtils.insertData("test", "test3", 0, 100, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ insertResult = TableModelUtils.insertData("test", "test4", 0, 200, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ TestUtils.assertDataEventuallyOnEnv(
+ receiverEnv,
+ "select count(*) from root.** where time <= 1",
+ "count(root.db.d4.at1),count(root.db.d2.at1),count(root.db.d3.at1),",
+ Collections.singleton("1,0,1,"));
+ TestUtils.assertDataEventuallyOnEnv(
+ receiverEnv,
+ "select count(*) from root.** where time >= 2",
+ "count(root.db.d4.at1),count(root.db.d2.at1),count(root.db.d3.at1),",
+ Collections.singleton("2,1,0,"));
+
+ TableModelUtils.assertData("test", "test2", 100, 200, receiverEnv);
+
+ TableModelUtils.assertData("test", "test3", 100, 200, receiverEnv);
+
+ TableModelUtils.assertData("test", "test4", 100, 200, receiverEnv);
+ }
+ }
+
+ @Test
+ public void testHistoryStartTimeAndEndTimeWorkingWithOrWithoutPattern() throws Exception {
+ final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+
+ final String receiverIp = receiverDataNode.getIp();
+ final int receiverPort = receiverDataNode.getPort();
+
+ boolean insertResult = true;
+
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+ if (!TestUtils.tryExecuteNonQueriesWithRetry(
+ senderEnv,
+ Arrays.asList(
+ "insert into root.db.d1 (time, at1)"
+ + " values (1, 1), (2, 2), (3, 3), (4, 4), (5, 5)",
+ "insert into root.db.d2 (time, at1)"
+ + " values (1, 1), (2, 2), (3, 3), (4, 4), (5, 5)",
+ "flush"))) {
+ return;
+ }
+
+ TableModelUtils.createDataBaseAndTable(senderEnv, "test1", "test");
+ insertResult = TableModelUtils.insertData("test", "test1", 0, 10, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ TableModelUtils.createDataBaseAndTable(senderEnv, "test2", "test");
+ insertResult = TableModelUtils.insertData("test", "test2", 0, 10, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+
+ // wait for flush to complete
+ if (!TestUtils.tryExecuteNonQueriesWithRetry(senderEnv, Collections.singletonList("flush"))) {
+ return;
+ }
+ Thread.sleep(10000);
+
+ final Map<String, String> extractorAttributes = new HashMap<>();
+ final Map<String, String> processorAttributes = new HashMap<>();
+ final Map<String, String> connectorAttributes = new HashMap<>();
+
+ extractorAttributes.put("extractor.capture.table", "true");
+ extractorAttributes.put("extractor.capture.tree", "true");
+ extractorAttributes.put("extractor.database-name", "test");
+ extractorAttributes.put("extractor.table-name", "test1");
+ extractorAttributes.put("extractor.pattern", "root.db.d1");
+ extractorAttributes.put("extractor.inclusion", "data.insert");
+ extractorAttributes.put("extractor.history.enable", "true");
+ // 1970-01-01T08:00:02+08:00
+ extractorAttributes.put("extractor.history.start-time", "2");
+ extractorAttributes.put("extractor.history.end-time", "3");
+
+ connectorAttributes.put("connector", "iotdb-thrift-connector");
+ connectorAttributes.put("connector.batch.enable", "false");
+ connectorAttributes.put("connector.ip", receiverIp);
+ connectorAttributes.put("connector.port", Integer.toString(receiverPort));
+
+ TSStatus status =
+ client.createPipe(
+ new TCreatePipeReq("p1", connectorAttributes)
+ .setExtractorAttributes(extractorAttributes)
+ .setProcessorAttributes(processorAttributes));
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("p1").getCode());
+
+ TestUtils.assertDataEventuallyOnEnv(
+ receiverEnv,
+ "select count(*) from root.**",
+ "count(root.db.d1.at1),",
+ Collections.singleton("2,"));
+
+ extractorAttributes.remove("extractor.table-name");
+ extractorAttributes.remove("extractor.pattern");
+ status =
+ client.createPipe(
+ new TCreatePipeReq("p2", connectorAttributes)
+ .setExtractorAttributes(extractorAttributes)
+ .setProcessorAttributes(processorAttributes));
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("p2").getCode());
+
+ TestUtils.assertDataEventuallyOnEnv(
+ receiverEnv,
+ "select count(*) from root.**",
+ "count(root.db.d1.at1),count(root.db.d2.at1),",
+ Collections.singleton("2,2,"));
+
+ TableModelUtils.assertData("test", "test1", 2, 4, receiverEnv);
+ TableModelUtils.assertData("test", "test2", 2, 4, receiverEnv);
+ }
+ }
+
+ @Test
+ public void testExtractorTimeRangeMatch() throws Exception {
+ final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+
+ final String receiverIp = receiverDataNode.getIp();
+ final int receiverPort = receiverDataNode.getPort();
+ boolean insertResult = true;
+
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+ // insert history data
+ if (!TestUtils.tryExecuteNonQueriesWithRetry(
+ senderEnv,
+ Arrays.asList(
+ "insert into root.db.d1 (time, at1)"
+ + " values (1, 1), (2, 2), (3, 3), (4, 4), (5, 5)",
+ "insert into root.db.d2 (time, at1)"
+ + " values (6, 6), (7, 7), (8, 8), (9, 9), (10, 10)",
+ "flush"))) {
+ return;
+ }
+
+ TableModelUtils.createDataBaseAndTable(senderEnv, "test1", "test");
+ insertResult = TableModelUtils.insertData("test", "test1", 0, 10, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ TableModelUtils.createDataBaseAndTable(senderEnv, "test2", "test");
+ insertResult = TableModelUtils.insertData("test", "test2", 0, 10, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ final Map<String, String> extractorAttributes = new HashMap<>();
+ final Map<String, String> processorAttributes = new HashMap<>();
+ final Map<String, String> connectorAttributes = new HashMap<>();
+
+ connectorAttributes.put("connector", "iotdb-thrift-connector");
+ connectorAttributes.put("connector.batch.enable", "false");
+ connectorAttributes.put("connector.ip", receiverIp);
+ connectorAttributes.put("connector.port", Integer.toString(receiverPort));
+
+ extractorAttributes.put("extractor.capture.table", "true");
+ extractorAttributes.put("extractor.capture.tree", "true");
+ extractorAttributes.put("source.inclusion", "data.insert");
+ extractorAttributes.put("source.start-time", "2");
+ extractorAttributes.put("source.end-time", "4");
+
+ final TSStatus status =
+ client.createPipe(
+ new TCreatePipeReq("p1", connectorAttributes)
+ .setExtractorAttributes(extractorAttributes)
+ .setProcessorAttributes(processorAttributes));
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+
+ TestUtils.assertDataEventuallyOnEnv(
+ receiverEnv,
+ "select count(*) from root.**",
+ "count(root.db.d1.at1),",
+ Collections.singleton("3,"));
+
+ TableModelUtils.assertCountData("test", "test1", 3, receiverEnv);
+
+ // Insert realtime data that overlapped with time range
+ if (!TestUtils.tryExecuteNonQueriesWithRetry(
+ senderEnv,
+ Arrays.asList(
+ "insert into root.db.d3 (time, at1)"
+ + " values (1, 1), (2, 2), (3, 3), (4, 4), (5, 5)",
+ "flush"))) {
+ return;
+ }
+ TableModelUtils.createDataBaseAndTable(senderEnv, "test3", "test");
+ insertResult = TableModelUtils.insertData("test", "test3", 0, 5, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ TestUtils.assertDataEventuallyOnEnv(
+ receiverEnv,
+ "select count(*) from root.**",
+ "count(root.db.d1.at1),count(root.db.d3.at1),",
+ Collections.singleton("3,3,"));
+
+ TableModelUtils.assertCountData("test", "test1", 3, receiverEnv);
+ TableModelUtils.assertCountData("test", "test3", 3, receiverEnv);
+
+ // Insert realtime data that does not overlap with time range
+ if (!TestUtils.tryExecuteNonQueriesWithRetry(
+ senderEnv,
+ Arrays.asList(
+ "insert into root.db.d4 (time, at1)"
+ + " values (6, 6), (7, 7), (8, 8), (9, 9), (10, 10)",
+ "flush"))) {
+ return;
+ }
+
+ TableModelUtils.createDataBaseAndTable(senderEnv, "test4", "test");
+ insertResult = TableModelUtils.insertData("test", "test4", 6, 10, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ TestUtils.assertDataAlwaysOnEnv(
+ receiverEnv,
+ "select count(*) from root.**",
+ "count(root.db.d1.at1),count(root.db.d3.at1),",
+ Collections.singleton("3,3,"));
+
+ TableModelUtils.assertCountData("test", "test1", 3, receiverEnv);
+ TableModelUtils.assertCountData("test", "test3", 3, receiverEnv);
+ }
+ }
+
+ @Test
+ public void testSourceStartTimeAndEndTimeWorkingWithOrWithoutPattern() throws Exception {
+ final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+
+ final String receiverIp = receiverDataNode.getIp();
+ final int receiverPort = receiverDataNode.getPort();
+ boolean insertResult = true;
+
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+ if (!TestUtils.tryExecuteNonQueriesWithRetry(
+ senderEnv,
+ Arrays.asList(
+ "insert into root.db.d1 (time, at1)"
+ + " values (1, 1), (2, 2), (3, 3), (4, 4), (5, 5)",
+ "insert into root.db.d2 (time, at1)"
+ + " values (1, 1), (2, 2), (3, 3), (4, 4), (5, 5)",
+ "flush"))) {
+ return;
+ }
+
+ TableModelUtils.createDataBaseAndTable(senderEnv, "test1", "test");
+ insertResult = TableModelUtils.insertData("test", "test1", 0, 5, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+
+ TableModelUtils.createDataBaseAndTable(senderEnv, "test2", "test");
+ insertResult = TableModelUtils.insertData("test", "test2", 0, 5, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ final Map<String, String> extractorAttributes = new HashMap<>();
+ final Map<String, String> processorAttributes = new HashMap<>();
+ final Map<String, String> connectorAttributes = new HashMap<>();
+
+ extractorAttributes.put("source.capture.table", "true");
+ extractorAttributes.put("source.capture.tree", "true");
+ extractorAttributes.put("source.pattern", "root.db.d1");
+ extractorAttributes.put("source.table-name", "test1");
+ extractorAttributes.put("source.inclusion", "data.insert");
+ extractorAttributes.put("source.start-time", "2");
+ // 1970-01-01T08:00:04+08:00
+ extractorAttributes.put("source.end-time", "4");
+
+ connectorAttributes.put("connector", "iotdb-thrift-connector");
+ connectorAttributes.put("connector.batch.enable", "false");
+ connectorAttributes.put("connector.ip", receiverIp);
+ connectorAttributes.put("connector.port", Integer.toString(receiverPort));
+
+ TSStatus status =
+ client.createPipe(
+ new TCreatePipeReq("p1", connectorAttributes)
+ .setExtractorAttributes(extractorAttributes)
+ .setProcessorAttributes(processorAttributes));
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("p1").getCode());
+
+ TestUtils.assertDataEventuallyOnEnv(
+ receiverEnv,
+ "select count(*) from root.**",
+ "count(root.db.d1.at1),",
+ Collections.singleton("3,"));
+ TableModelUtils.assertCountData("test", "test1", 3, receiverEnv);
+
+ extractorAttributes.remove("source.pattern");
+ extractorAttributes.remove("source.table-name");
+ status =
+ client.createPipe(
+ new TCreatePipeReq("p2", connectorAttributes)
+ .setExtractorAttributes(extractorAttributes)
+ .setProcessorAttributes(processorAttributes));
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("p2").getCode());
+
+ TestUtils.assertDataEventuallyOnEnv(
+ receiverEnv,
+ "select count(*) from root.**",
+ "count(root.db.d1.at1),count(root.db.d2.at1),",
+ Collections.singleton("3,3,"));
+
+ TableModelUtils.assertCountData("test", "test1", 3, receiverEnv);
+ TableModelUtils.assertCountData("test", "test2", 3, receiverEnv);
+ }
+ }
+
+ @Test
+ public void testHistoryLooseRange() throws Exception {
+ final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+
+ final String receiverIp = receiverDataNode.getIp();
+ final int receiverPort = receiverDataNode.getPort();
+ boolean insertResult = true;
+
+ TableModelUtils.createDataBaseAndTable(senderEnv, "test1", "test");
+ insertResult = TableModelUtils.insertData("test", "test1", 0, 2, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+
+ TableModelUtils.createDataBaseAndTable(senderEnv, "test2", "test");
+ insertResult = TableModelUtils.insertData("test", "test2", 0, 2, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+ if (!TestUtils.tryExecuteNonQueriesWithRetry(
+ senderEnv,
+ Arrays.asList(
+ // TsFile 1, extracted without parse
+ "insert into root.db.d1 (time, at1, at2)" + " values (1, 1, 2), (2, 3, 4)",
+ // TsFile 2, not extracted because pattern not overlapped
+ "insert into root.db1.d1 (time, at1, at2)" + " values (1, 1, 2), (2, 3, 4)",
+ "flush"))) {
+ return;
+ }
+
+ if (!TestUtils.tryExecuteNonQueriesWithRetry(
+ senderEnv,
+ Arrays.asList(
+ // TsFile 3, not extracted because time range not overlapped
+ "insert into root.db.d1 (time, at1, at2)" + " values (3, 1, 2), (4, 3, 4)",
+ "flush"))) {
+ return;
+ }
+
+ insertResult = TableModelUtils.insertData("test", "test1", 2, 5, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+
+ // wait for flush to complete
+ if (!TestUtils.tryExecuteNonQueriesWithRetry(senderEnv, Collections.singletonList("flush"))) {
+ return;
+ }
+ Thread.sleep(10000);
+
+ final Map<String, String> extractorAttributes = new HashMap<>();
+ final Map<String, String> processorAttributes = new HashMap<>();
+ final Map<String, String> connectorAttributes = new HashMap<>();
+
+ extractorAttributes.put("source.capture.table", "true");
+ extractorAttributes.put("source.capture.tree", "true");
+ extractorAttributes.put("source.table-name", "test1");
+ extractorAttributes.put("source.path", "root.db.d1.at1");
+ extractorAttributes.put("source.inclusion", "data.insert");
+ extractorAttributes.put("source.history.start-time", "1");
+ extractorAttributes.put("source.history.end-time", "2");
+ extractorAttributes.put("source.history.loose-range", "time, path");
+
+ connectorAttributes.put("connector", "iotdb-thrift-connector");
+ connectorAttributes.put("connector.batch.enable", "false");
+ connectorAttributes.put("connector.ip", receiverIp);
+ connectorAttributes.put("connector.port", Integer.toString(receiverPort));
+
+ TSStatus status =
+ client.createPipe(
+ new TCreatePipeReq("p1", connectorAttributes)
+ .setExtractorAttributes(extractorAttributes)
+ .setProcessorAttributes(processorAttributes));
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("p1").getCode());
+
+ TestUtils.assertDataEventuallyOnEnv(
+ receiverEnv,
+ "select count(*) from root.** group by level=0",
+ "count(root.*.*.*),",
+ Collections.singleton("4,"));
+
+ TableModelUtils.assertCountData("test", "test1", 3, receiverEnv);
+ }
+ }
+
+ @Test
+ public void testRealtimeLooseRange() throws Exception {
+ final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+ final String receiverIp = receiverDataNode.getIp();
+ final int receiverPort = receiverDataNode.getPort();
+
+ boolean insertResult = true;
+ insertResult = TableModelUtils.insertData("test", "test2", 0, 20, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ final Map<String, String> extractorAttributes = new HashMap<>();
+ final Map<String, String> processorAttributes = new HashMap<>();
+ final Map<String, String> connectorAttributes = new HashMap<>();
+
+ extractorAttributes.put("source.capture.table", "true");
+ extractorAttributes.put("source.capture.tree", "true");
+ extractorAttributes.put("source.table-name", "test1");
+ extractorAttributes.put("source.path", "root.db.d1.at1");
+ extractorAttributes.put("source.inclusion", "data.insert");
+ extractorAttributes.put("source.realtime.loose-range", "time, path");
+ extractorAttributes.put("source.start-time", "2");
+ extractorAttributes.put("source.end-time", "10");
+ extractorAttributes.put("source.realtime.mode", "batch");
+
+ connectorAttributes.put("connector", "iotdb-thrift-connector");
+ connectorAttributes.put("connector.batch.enable", "false");
+ connectorAttributes.put("connector.ip", receiverIp);
+ connectorAttributes.put("connector.port", Integer.toString(receiverPort));
+
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+ TSStatus status =
+ client.createPipe(
+ new TCreatePipeReq("p1", connectorAttributes)
+ .setExtractorAttributes(extractorAttributes)
+ .setProcessorAttributes(processorAttributes));
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("p1").getCode());
+
+ if (!TestUtils.tryExecuteNonQueriesWithRetry(
+ senderEnv,
+ Arrays.asList(
+ "insert into root.db.d1 (time, at1, at2)" + " values (1, 1, 2), (3, 3, 4)",
+ "flush"))) {
+ return;
+ }
+
+ if (!insertResult) {
+ return;
+ }
+
+ if (!TestUtils.tryExecuteNonQueriesWithRetry(
+ senderEnv,
+ Arrays.asList(
+ "insert into root.db.d1 (time, at1)" + " values (5, 1), (16, 3)",
+ "insert into root.db.d1 (time, at1, at2)" + " values (5, 1, 2), (6, 3, 4)",
+ "flush"))) {
+ return;
+ }
+
+ TestUtils.assertDataEventuallyOnEnv(
+ receiverEnv,
+ "select count(at1) from root.db.d1 where time >= 2 and time <= 10",
+ new HashMap<String, String>() {
+ {
+ put("count(root.db.d1.at1)", "3");
+ }
+ });
+
+ insertResult = TableModelUtils.insertData("test", "test1", 10, 20, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ insertResult = TableModelUtils.insertData("test", "test2", 10, 20, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+
+ TableModelUtils.assertCountData("test", "test1", 20, receiverEnv);
+ }
+ }
+
+ @Test
+ public void testTableModeSQLSupportNowFunc() {
+ final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+
+ final String receiverIp = receiverDataNode.getIp();
+ final int receiverPort = receiverDataNode.getPort();
+
+ final String p1 =
+ String.format(
+ "create pipe p1"
+ + " with extractor ("
+ + "'capture.table'='true',"
+ + "'extractor.history.enable'='true',"
+ + "'source.start-time'='now',"
+ + "'source.end-time'='now',"
+ + "'source.history.start-time'='now',"
+ + "'source.history.end-time'='now')"
+ + " with connector ("
+ + "'connector'='iotdb-thrift-connector',"
+ + "'connector.ip'='%s',"
+ + "'connector.port'='%s',"
+ + "'connector.batch.enable'='false')",
+ receiverIp, receiverPort);
+ try (final Connection connection = senderEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ final Statement statement = connection.createStatement()) {
+ statement.execute(p1);
+ } catch (final SQLException e) {
+ fail(e.getMessage());
+ }
+
+ final String p2 =
+ String.format(
+ "create pipe p2"
+ + " with extractor ("
+ + "'capture.table'='true',"
+ + "'extractor.history.enable'='true',"
+ + "'start-time'='now',"
+ + "'end-time'='now',"
+ + "'history.start-time'='now',"
+ + "'history.end-time'='now')"
+ + " with connector ("
+ + "'connector'='iotdb-thrift-connector',"
+ + "'connector.ip'='%s',"
+ + "'connector.port'='%s',"
+ + "'connector.batch.enable'='false')",
+ receiverIp, receiverPort);
+ try (final Connection connection = senderEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ final Statement statement = connection.createStatement()) {
+ statement.execute(p2);
+ } catch (final SQLException e) {
+ fail(e.getMessage());
+ }
+
+ final String p3 =
+ String.format(
+ "create pipe p3"
+ + " with extractor ("
+ + "'capture.table'='true',"
+ + "'extractor.history.enable'='true',"
+ + "'extractor.start-time'='now',"
+ + "'extractor.end-time'='now',"
+ + "'extractor.history.start-time'='now',"
+ + "'extractor.history.end-time'='now')"
+ + " with connector ("
+ + "'connector'='iotdb-thrift-connector',"
+ + "'connector.ip'='%s',"
+ + "'connector.port'='%s',"
+ + "'connector.batch.enable'='false')",
+ receiverIp, receiverPort);
+ try (final Connection connection = senderEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ final Statement statement = connection.createStatement()) {
+ statement.execute(p3);
+ } catch (final SQLException e) {
+ fail(e.getMessage());
+ }
+
+ String alterP3 =
+ "alter pipe p3"
+ + " modify extractor ("
+ + "'history.enable'='true',"
+ + "'start-time'='now',"
+ + "'end-time'='now',"
+ + "'history.start-time'='now',"
+ + "'history.end-time'='now')";
+ try (final Connection connection = senderEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ final Statement statement = connection.createStatement()) {
+ statement.execute(alterP3);
+ } catch (final SQLException e) {
+ fail(e.getMessage());
+ }
+
+ alterP3 =
+ "alter pipe p3"
+ + " modify extractor ("
+ + "'extractor.history.enable'='true',"
+ + "'extractor.start-time'='now',"
+ + "'extractor.end-time'='now',"
+ + "'extractor.history.start-time'='now',"
+ + "'extractor.history.end-time'='now')";
+ try (final Connection connection = senderEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ final Statement statement = connection.createStatement()) {
+ statement.execute(alterP3);
+ } catch (final SQLException e) {
+ fail(e.getMessage());
+ }
+
+ alterP3 =
+ "alter pipe p3"
+ + " modify source ("
+ + "'extractor.history.enable'='true',"
+ + "'source.start-time'='now',"
+ + "'source.end-time'='now',"
+ + "'source.history.start-time'='now',"
+ + "'source.history.end-time'='now')";
+ try (final Connection connection = senderEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ final Statement statement = connection.createStatement()) {
+ statement.execute(alterP3);
+ } catch (final SQLException e) {
+ fail(e.getMessage());
+ }
+ }
+
+ private void assertTimeseriesCountOnReceiver(BaseEnv receiverEnv, int count) {
+ TestUtils.assertDataEventuallyOnEnv(
+ receiverEnv, "count timeseries", "count(timeseries),", Collections.singleton(count + ","));
+ }
+}
diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/tablemodel/IoTDBPipeLifeCycleIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/tablemodel/IoTDBPipeLifeCycleIT.java
new file mode 100644
index 00000000000..3c52a1de726
--- /dev/null
+++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/tablemodel/IoTDBPipeLifeCycleIT.java
@@ -0,0 +1,642 @@
+/*
+ * 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.pipe.it.tablemodel;
+
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.client.sync.SyncConfigNodeIServiceClient;
+import org.apache.iotdb.confignode.rpc.thrift.TCreatePipeReq;
+import org.apache.iotdb.db.it.utils.TestUtils;
+import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper;
+import org.apache.iotdb.it.framework.IoTDBTestRunner;
+import org.apache.iotdb.itbase.category.MultiClusterIT2TableModel;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+
+@RunWith(IoTDBTestRunner.class)
+@Category({MultiClusterIT2TableModel.class})
+public class IoTDBPipeLifeCycleIT extends AbstractPipeTableModelTestIT {
+ @Test
+ public void testLifeCycleWithHistoryEnabled() throws Exception {
+ final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+
+ final String receiverIp = receiverDataNode.getIp();
+ final int receiverPort = receiverDataNode.getPort();
+ boolean insertResult = true;
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+
+ TableModelUtils.createDataBaseAndTable(senderEnv, "test", "test");
+ insertResult = TableModelUtils.insertData("test", "test", 0, 100, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ final Map<String, String> extractorAttributes = new HashMap<>();
+ final Map<String, String> processorAttributes = new HashMap<>();
+ final Map<String, String> connectorAttributes = new HashMap<>();
+
+ extractorAttributes.put("capture.table", "true");
+
+ connectorAttributes.put("connector", "iotdb-thrift-connector");
+ connectorAttributes.put("connector.batch.enable", "false");
+ connectorAttributes.put("connector.ip", receiverIp);
+ connectorAttributes.put("connector.port", Integer.toString(receiverPort));
+
+ final TSStatus status =
+ client.createPipe(
+ new TCreatePipeReq("p1", connectorAttributes)
+ .setExtractorAttributes(extractorAttributes)
+ .setProcessorAttributes(processorAttributes));
+
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("p1").getCode());
+
+ insertResult = TableModelUtils.insertData("test", "test", 100, 200, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ TableModelUtils.assertCountData("test", "test", 200, receiverEnv);
+
+ insertResult = TableModelUtils.insertData("test", "test", 200, 300, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ TableModelUtils.assertCountData("test", "test", 300, receiverEnv);
+
+ insertResult = TableModelUtils.insertData("test", "test", 300, 400, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ TableModelUtils.assertCountData("test", "test", 400, receiverEnv);
+
+ insertResult = TableModelUtils.insertData("test", "test", 400, 500, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ TableModelUtils.assertCountData("test", "test", 500, receiverEnv);
+ }
+ }
+
+ @Test
+ public void testLifeCycleWithHistoryDisabled() throws Exception {
+ final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+
+ final String receiverIp = receiverDataNode.getIp();
+ final int receiverPort = receiverDataNode.getPort();
+ boolean insertResult = true;
+
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+
+ TableModelUtils.createDataBaseAndTable(senderEnv, "test", "test");
+ insertResult = TableModelUtils.insertData("test", "test", 0, 100, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+
+ // wait for flush to complete
+ if (!TestUtils.tryExecuteNonQueriesWithRetry(senderEnv, Collections.singletonList("flush"))) {
+ return;
+ }
+ Thread.sleep(10000);
+
+ final Map<String, String> extractorAttributes = new HashMap<>();
+ final Map<String, String> processorAttributes = new HashMap<>();
+ final Map<String, String> connectorAttributes = new HashMap<>();
+
+ extractorAttributes.put("capture.table", "true");
+ extractorAttributes.put("extractor.inclusion", "data.insert");
+ extractorAttributes.put("extractor.inclusion.exclusion", "");
+
+ extractorAttributes.put("extractor.history.enable", "false");
+ // start-time and end-time should not work
+ extractorAttributes.put("extractor.history.start-time", "0");
+ extractorAttributes.put("extractor.history.end-time", "50");
+
+ connectorAttributes.put("connector", "iotdb-thrift-connector");
+ connectorAttributes.put("connector.batch.enable", "false");
+ connectorAttributes.put("connector.ip", receiverIp);
+ connectorAttributes.put("connector.port", Integer.toString(receiverPort));
+
+ final TSStatus status =
+ client.createPipe(
+ new TCreatePipeReq("p1", connectorAttributes)
+ .setExtractorAttributes(extractorAttributes)
+ .setProcessorAttributes(processorAttributes));
+
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("p1").getCode());
+
+ TableModelUtils.createDataBaseAndTable(senderEnv, "test1", "test1");
+ insertResult = TableModelUtils.insertData("test1", "test1", 0, 100, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+
+ TableModelUtils.assertCountData("test1", "test1", 100, receiverEnv);
+
+ TableModelUtils.createDataBaseAndTable(senderEnv, "test2", "test2");
+ insertResult = TableModelUtils.insertData("test2", "test2", 0, 100, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ TableModelUtils.assertCountData("test1", "test1", 100, receiverEnv);
+ }
+ }
+
+ @Test
+ public void testLifeCycleLogMode() throws Exception {
+ final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+
+ final String receiverIp = receiverDataNode.getIp();
+ final int receiverPort = receiverDataNode.getPort();
+ boolean insertResult = true;
+
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+
+ final Map<String, String> extractorAttributes = new HashMap<>();
+ final Map<String, String> processorAttributes = new HashMap<>();
+ final Map<String, String> connectorAttributes = new HashMap<>();
+
+ extractorAttributes.put("capture.table", "true");
+ extractorAttributes.put("extractor.mode", "forced-log");
+
+ connectorAttributes.put("connector", "iotdb-thrift-connector");
+ connectorAttributes.put("connector.batch.enable", "false");
+ connectorAttributes.put("connector.ip", receiverIp);
+ connectorAttributes.put("connector.port", Integer.toString(receiverPort));
+
+ final TSStatus status =
+ client.createPipe(
+ new TCreatePipeReq("p1", connectorAttributes)
+ .setExtractorAttributes(extractorAttributes)
+ .setProcessorAttributes(processorAttributes));
+
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("p1").getCode());
+
+ TableModelUtils.createDataBaseAndTable(senderEnv, "test", "test");
+ insertResult = TableModelUtils.insertData("test", "test", 0, 100, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ TableModelUtils.assertCountData("test", "test", 100, senderEnv);
+
+ TableModelUtils.createDataBaseAndTable(senderEnv, "test1", "test1");
+ insertResult = TableModelUtils.insertData("test1", "test1", 0, 100, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ TableModelUtils.assertCountData("test1", "test1", 100, senderEnv);
+
+ TableModelUtils.createDataBaseAndTable(senderEnv, "test2", "test2");
+ insertResult = TableModelUtils.insertData("test2", "test2", 0, 100, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ TableModelUtils.assertCountData("test2", "test2", 100, senderEnv);
+ }
+ }
+
+ @Test
+ public void testLifeCycleFileMode() throws Exception {
+ final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+
+ final String receiverIp = receiverDataNode.getIp();
+ final int receiverPort = receiverDataNode.getPort();
+ boolean insertResult = true;
+
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+
+ final Map<String, String> extractorAttributes = new HashMap<>();
+ final Map<String, String> processorAttributes = new HashMap<>();
+ final Map<String, String> connectorAttributes = new HashMap<>();
+
+ extractorAttributes.put("capture.table", "true");
+ extractorAttributes.put("mode.streaming", "false");
+
+ connectorAttributes.put("connector", "iotdb-thrift-connector");
+ connectorAttributes.put("connector.batch.enable", "false");
+ connectorAttributes.put("connector.ip", receiverIp);
+ connectorAttributes.put("connector.port", Integer.toString(receiverPort));
+
+ final TSStatus status =
+ client.createPipe(
+ new TCreatePipeReq("p1", connectorAttributes)
+ .setExtractorAttributes(extractorAttributes)
+ .setProcessorAttributes(processorAttributes));
+
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("p1").getCode());
+
+ TableModelUtils.createDataBaseAndTable(senderEnv, "test", "test");
+ insertResult = TableModelUtils.insertData("test", "test", 0, 100, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ TableModelUtils.assertCountData("test", "test", 100, receiverEnv);
+
+ TableModelUtils.createDataBaseAndTable(senderEnv, "test1", "test1");
+ insertResult = TableModelUtils.insertData("test1", "test1", 0, 100, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ TableModelUtils.assertCountData("test1", "test1", 100, receiverEnv);
+ }
+ }
+
+ @Test
+ public void testLifeCycleHybridMode() throws Exception {
+ final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+
+ final String receiverIp = receiverDataNode.getIp();
+ final int receiverPort = receiverDataNode.getPort();
+ boolean insertResult = true;
+
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+
+ TableModelUtils.createDataBaseAndTable(senderEnv, "test", "test");
+ insertResult = TableModelUtils.insertData("test", "test", 0, 100, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+
+ final Map<String, String> extractorAttributes = new HashMap<>();
+ final Map<String, String> processorAttributes = new HashMap<>();
+ final Map<String, String> connectorAttributes = new HashMap<>();
+
+ extractorAttributes.put("capture.table", "true");
+ extractorAttributes.put("extractor.mode", "hybrid");
+
+ connectorAttributes.put("connector", "iotdb-thrift-connector");
+ connectorAttributes.put("connector.batch.enable", "false");
+ connectorAttributes.put("connector.ip", receiverIp);
+ connectorAttributes.put("connector.port", Integer.toString(receiverPort));
+
+ final TSStatus status =
+ client.createPipe(
+ new TCreatePipeReq("p1", connectorAttributes)
+ .setExtractorAttributes(extractorAttributes)
+ .setProcessorAttributes(processorAttributes));
+
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("p1").getCode());
+
+ insertResult = TableModelUtils.insertData("test", "test", 100, 200, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ TableModelUtils.assertCountData("test", "test", 200, receiverEnv);
+
+ insertResult = TableModelUtils.insertData("test", "test", 200, 300, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ TableModelUtils.assertCountData("test", "test", 300, receiverEnv);
+ }
+ }
+
+ @Test
+ public void testLifeCycleWithClusterRestart() throws Exception {
+ final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+
+ final String receiverIp = receiverDataNode.getIp();
+ final int receiverPort = receiverDataNode.getPort();
+ boolean insertResult = true;
+
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+
+ TableModelUtils.createDataBaseAndTable(senderEnv, "test", "test");
+ insertResult = TableModelUtils.insertData("test", "test", 0, 100, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ final Map<String, String> extractorAttributes = new HashMap<>();
+ final Map<String, String> processorAttributes = new HashMap<>();
+ final Map<String, String> connectorAttributes = new HashMap<>();
+
+ extractorAttributes.put("capture.table", "true");
+
+ connectorAttributes.put("connector", "iotdb-thrift-connector");
+ connectorAttributes.put("connector.batch.enable", "false");
+ connectorAttributes.put("connector.ip", receiverIp);
+ connectorAttributes.put("connector.port", Integer.toString(receiverPort));
+
+ final TSStatus status =
+ client.createPipe(
+ new TCreatePipeReq("p1", connectorAttributes)
+ .setExtractorAttributes(extractorAttributes)
+ .setProcessorAttributes(processorAttributes));
+
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("p1").getCode());
+
+ insertResult = TableModelUtils.insertData("test", "test", 100, 200, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ TableModelUtils.assertCountData("test", "test", 200, receiverEnv);
+ }
+
+ try {
+ TestUtils.restartCluster(senderEnv);
+ TestUtils.restartCluster(receiverEnv);
+ } catch (final Throwable e) {
+ e.printStackTrace();
+ return;
+ }
+
+ try (final SyncConfigNodeIServiceClient ignored =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+
+ insertResult = TableModelUtils.insertData("test", "test", 200, 300, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ TableModelUtils.assertCountData("test", "test", 300, receiverEnv);
+ }
+ }
+
+ @Test
+ public void testReceiverRestartWhenTransferring() throws Exception {
+ final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+
+ final String receiverIp = receiverDataNode.getIp();
+ final int receiverPort = receiverDataNode.getPort();
+ boolean insertResult = true;
+
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+
+ TableModelUtils.createDataBaseAndTable(senderEnv, "test", "test");
+ insertResult = TableModelUtils.insertData("test", "test", 0, 100, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+
+ final Map<String, String> extractorAttributes = new HashMap<>();
+ final Map<String, String> processorAttributes = new HashMap<>();
+ final Map<String, String> connectorAttributes = new HashMap<>();
+
+ extractorAttributes.put("capture.table", "true");
+
+ connectorAttributes.put("connector", "iotdb-thrift-connector");
+ connectorAttributes.put("connector.batch.enable", "false");
+ connectorAttributes.put("connector.ip", receiverIp);
+ connectorAttributes.put("connector.port", Integer.toString(receiverPort));
+
+ final TSStatus status =
+ client.createPipe(
+ new TCreatePipeReq("p1", connectorAttributes)
+ .setExtractorAttributes(extractorAttributes)
+ .setProcessorAttributes(processorAttributes));
+
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("p1").getCode());
+
+ final AtomicInteger succeedNum = new AtomicInteger(0);
+ final Thread t =
+ new Thread(
+ () -> {
+ try {
+ for (int i = 100; i < 200; ++i) {
+ if (TableModelUtils.insertDataNotThrowError(
+ "test", "test", i, i + 1, senderEnv)) {
+ succeedNum.incrementAndGet();
+ }
+ Thread.sleep(100);
+ }
+ } catch (InterruptedException ignored) {
+ }
+ });
+ t.start();
+
+ try {
+ TestUtils.restartCluster(receiverEnv);
+ } catch (final Throwable e) {
+ e.printStackTrace();
+ try {
+ t.interrupt();
+ t.join();
+ } catch (Throwable ignored) {
+ }
+ return;
+ }
+
+ t.join();
+ client.stopPipe("p1");
+ client.startPipe("p1");
+ if (!TestUtils.tryExecuteNonQueryWithRetry(senderEnv, "flush")) {
+ return;
+ }
+ TableModelUtils.assertCountData("test", "test", 100 + succeedNum.get(), receiverEnv);
+ }
+ }
+
+ @Test
+ public void testReceiverAlreadyHaveTimeSeries() throws Exception {
+ final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+
+ final String receiverIp = receiverDataNode.getIp();
+ final int receiverPort = receiverDataNode.getPort();
+ boolean insertResult = true;
+
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+
+ TableModelUtils.createDataBaseAndTable(receiverEnv, "test", "test");
+ TableModelUtils.createDataBaseAndTable(senderEnv, "test", "test");
+ insertResult = TableModelUtils.insertData("test", "test", 0, 100, receiverEnv);
+ if (!insertResult) {
+ return;
+ }
+ final Map<String, String> extractorAttributes = new HashMap<>();
+ final Map<String, String> processorAttributes = new HashMap<>();
+ final Map<String, String> connectorAttributes = new HashMap<>();
+
+ extractorAttributes.put("capture.table", "true");
+
+ connectorAttributes.put("connector", "iotdb-thrift-connector");
+ connectorAttributes.put("connector.batch.enable", "false");
+ connectorAttributes.put("connector.ip", receiverIp);
+ connectorAttributes.put("connector.port", Integer.toString(receiverPort));
+
+ final TSStatus status =
+ client.createPipe(
+ new TCreatePipeReq("p1", connectorAttributes)
+ .setExtractorAttributes(extractorAttributes)
+ .setProcessorAttributes(processorAttributes));
+
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("p1").getCode());
+
+ insertResult = TableModelUtils.insertData("test", "test", 100, 200, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ TableModelUtils.assertCountData("test", "test", 200, receiverEnv);
+
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.stopPipe("p1").getCode());
+
+ insertResult = TableModelUtils.insertData("test", "test", 200, 300, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ TableModelUtils.assertCountData("test", "test", 200, receiverEnv);
+ }
+ }
+
+ @Test
+ public void testDoubleLiving() throws Exception {
+ boolean insertResult = true;
+ // Double living is two clusters with pipes connecting each other.
+ final DataNodeWrapper senderDataNode = senderEnv.getDataNodeWrapper(0);
+ final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+
+ final String senderIp = senderDataNode.getIp();
+ final int senderPort = senderDataNode.getPort();
+ final String receiverIp = receiverDataNode.getIp();
+ final int receiverPort = receiverDataNode.getPort();
+
+ if (!TestUtils.tryExecuteNonQueryWithRetry(senderEnv, "flush")) {
+ return;
+ }
+
+ TableModelUtils.createDataBaseAndTable(senderEnv, "test", "test");
+ insertResult = TableModelUtils.insertData("test", "test", 0, 100, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+ final Map<String, String> extractorAttributes = new HashMap<>();
+ final Map<String, String> processorAttributes = new HashMap<>();
+ final Map<String, String> connectorAttributes = new HashMap<>();
+
+ // Add this property to avoid to make self cycle.
+ extractorAttributes.put("capture.table", "true");
+ connectorAttributes.put("source.forwarding-pipe-requests", "false");
+ connectorAttributes.put("connector", "iotdb-thrift-connector");
+ connectorAttributes.put("connector.batch.enable", "false");
+ connectorAttributes.put("connector.ip", receiverIp);
+ connectorAttributes.put("connector.port", Integer.toString(receiverPort));
+
+ final TSStatus status =
+ client.createPipe(
+ new TCreatePipeReq("p1", connectorAttributes)
+ .setExtractorAttributes(extractorAttributes)
+ .setProcessorAttributes(processorAttributes));
+
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("p1").getCode());
+ }
+
+ if (!TestUtils.tryExecuteNonQueryWithRetry(senderEnv, "flush")) {
+ return;
+ }
+
+ insertResult = TableModelUtils.insertData("test", "test", 100, 200, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+
+ insertResult = TableModelUtils.insertData("test", "test", 200, 300, receiverEnv);
+ if (!insertResult) {
+ return;
+ }
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) receiverEnv.getLeaderConfigNodeConnection()) {
+ final Map<String, String> extractorAttributes = new HashMap<>();
+ final Map<String, String> processorAttributes = new HashMap<>();
+ final Map<String, String> connectorAttributes = new HashMap<>();
+
+ // Add this property to avoid to make self cycle.
+ extractorAttributes.put("capture.table", "true");
+ extractorAttributes.put("capture.tree", "true");
+ connectorAttributes.put("source.forwarding-pipe-requests", "false");
+ connectorAttributes.put("connector", "iotdb-thrift-connector");
+ connectorAttributes.put("connector.batch.enable", "false");
+ connectorAttributes.put("connector.ip", senderIp);
+ connectorAttributes.put("connector.port", Integer.toString(senderPort));
+
+ final TSStatus status =
+ client.createPipe(
+ new TCreatePipeReq("p1", connectorAttributes)
+ .setExtractorAttributes(extractorAttributes)
+ .setProcessorAttributes(processorAttributes));
+
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("p1").getCode());
+ }
+
+ if (!TestUtils.tryExecuteNonQueryWithRetry(receiverEnv, "flush")) {
+ return;
+ }
+ insertResult = TableModelUtils.insertData("test", "test", 300, 400, receiverEnv);
+ if (!insertResult) {
+ return;
+ }
+ TableModelUtils.assertData("test", "test", 0, 400, receiverEnv);
+
+ try {
+ TestUtils.restartCluster(senderEnv);
+ TestUtils.restartCluster(receiverEnv);
+ } catch (final Throwable e) {
+ e.printStackTrace();
+ return;
+ }
+
+ insertResult = TableModelUtils.insertData("test", "test", 400, 500, receiverEnv);
+ if (!insertResult) {
+ return;
+ }
+
+ TableModelUtils.assertData("test", "test", 0, 500, receiverEnv);
+ }
+}
diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/tablemodel/IoTDBPipeNullValueIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/tablemodel/IoTDBPipeNullValueIT.java
new file mode 100644
index 00000000000..fb63d43d8d3
--- /dev/null
+++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/tablemodel/IoTDBPipeNullValueIT.java
@@ -0,0 +1,169 @@
+/*
+ * 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.pipe.it.tablemodel;
+
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.client.sync.SyncConfigNodeIServiceClient;
+import org.apache.iotdb.confignode.rpc.thrift.TCreatePipeReq;
+import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper;
+import org.apache.iotdb.it.framework.IoTDBTestRunner;
+import org.apache.iotdb.itbase.category.MultiClusterIT2TableModel;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import java.util.HashMap;
+import java.util.Map;
+
+@RunWith(IoTDBTestRunner.class)
+@Category({MultiClusterIT2TableModel.class})
+public class IoTDBPipeNullValueIT extends AbstractPipeTableModelTestIT {
+
+ private enum InsertType {
+ SESSION_INSERT_RECORD,
+ SESSION_INSERT_TABLET,
+ SQL_INSERT,
+ }
+
+ private void testInsertNullValueTemplate(
+ final InsertType insertType, final boolean withParsing, final String realtime)
+ throws Exception {
+ final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+
+ final String receiverIp = receiverDataNode.getIp();
+ final int receiverPort = receiverDataNode.getPort();
+
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+ final Map<String, String> extractorAttributes = new HashMap<>();
+ final Map<String, String> processorAttributes = new HashMap<>();
+ final Map<String, String> connectorAttributes = new HashMap<>();
+
+ TableModelUtils.createDataBaseAndTable(senderEnv, "test", "test");
+
+ if (insertType == InsertType.SESSION_INSERT_TABLET) {
+ TableModelUtils.insertDataByTablet("test", "test", 0, 200, senderEnv, true);
+ } else if (insertType == InsertType.SQL_INSERT) {
+ TableModelUtils.insertData("test", "test", 0, 200, senderEnv, true);
+ }
+
+ connectorAttributes.put("connector", "iotdb-thrift-connector");
+ connectorAttributes.put("connector.ip", receiverIp);
+ connectorAttributes.put("connector.port", Integer.toString(receiverPort));
+
+ extractorAttributes.put("capture.table", "true");
+ extractorAttributes.put("realtime-mode", realtime);
+ if (withParsing) {
+ extractorAttributes.put("start-time", "150");
+ extractorAttributes.put("end-time", "249");
+ extractorAttributes.put("database-name", "test");
+ extractorAttributes.put("table-name", "test");
+ }
+
+ final TSStatus status =
+ client.createPipe(
+ new TCreatePipeReq("test", connectorAttributes)
+ .setExtractorAttributes(extractorAttributes)
+ .setProcessorAttributes(processorAttributes));
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+ }
+
+ if (insertType == InsertType.SESSION_INSERT_TABLET) {
+ TableModelUtils.insertDataByTablet("test", "test", 200, 400, senderEnv, true);
+ } else if (insertType == InsertType.SQL_INSERT) {
+ TableModelUtils.insertData("test", "test", 200, 400, senderEnv, true);
+ }
+
+ if (withParsing) {
+ TableModelUtils.assertCountData("test", "test", 100, receiverEnv);
+ return;
+ }
+ TableModelUtils.assertCountData("test", "test", 400, receiverEnv);
+ }
+
+ // ---------------------- //
+ // Scenario 1: SQL Insert //
+ // ---------------------- //
+ @Test
+ public void testSQLInsertWithParsingForcedLog() throws Exception {
+ testInsertNullValueTemplate(InsertType.SQL_INSERT, true, "forced-log");
+ }
+
+ @Test
+ public void testSQLInsertWithoutParsingForcedLog() throws Exception {
+ testInsertNullValueTemplate(InsertType.SQL_INSERT, false, "forced-log");
+ }
+
+ @Test
+ public void testSQLInsertWithParsingFile() throws Exception {
+ testInsertNullValueTemplate(InsertType.SQL_INSERT, true, "file");
+ }
+
+ @Test
+ public void testSQLInsertWithoutParsingFile() throws Exception {
+ testInsertNullValueTemplate(InsertType.SQL_INSERT, false, "file");
+ }
+
+ @Test
+ public void testSQLInsertWithParsingStream() throws Exception {
+ testInsertNullValueTemplate(InsertType.SQL_INSERT, true, "stream");
+ }
+
+ @Test
+ public void testSQLInsertWithoutParsingStream() throws Exception {
+ testInsertNullValueTemplate(InsertType.SQL_INSERT, false, "stream");
+ }
+
+ // --------------------------------- //
+ // Scenario 2: Session Insert Tablet //
+ // --------------------------------- //
+ @Test
+ public void testSessionInsertTabletWithParsingForcedLog() throws Exception {
+ testInsertNullValueTemplate(InsertType.SESSION_INSERT_TABLET, true, "forced-log");
+ }
+
+ @Test
+ public void testSessionInsertTabletWithoutParsingForcedLog() throws Exception {
+ testInsertNullValueTemplate(InsertType.SESSION_INSERT_TABLET, false, "forced-log");
+ }
+
+ @Test
+ public void testSessionInsertTabletWithParsingFile() throws Exception {
+ testInsertNullValueTemplate(InsertType.SESSION_INSERT_TABLET, true, "file");
+ }
+
+ @Test
+ public void testSessionInsertTabletWithoutParsingFile() throws Exception {
+ testInsertNullValueTemplate(InsertType.SESSION_INSERT_TABLET, false, "file");
+ }
+
+ @Test
+ public void testSessionInsertTabletWithParsingStream() throws Exception {
+ testInsertNullValueTemplate(InsertType.SESSION_INSERT_TABLET, true, "stream");
+ }
+
+ @Test
+ public void testSessionInsertTabletWithoutParsingStream() throws Exception {
+ testInsertNullValueTemplate(InsertType.SESSION_INSERT_TABLET, false, "stream");
+ }
+}
diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/tablemodel/IoTDBPipeProtocolIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/tablemodel/IoTDBPipeProtocolIT.java
new file mode 100644
index 00000000000..fc78f59d826
--- /dev/null
+++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/tablemodel/IoTDBPipeProtocolIT.java
@@ -0,0 +1,466 @@
+/*
+ * 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.pipe.it.tablemodel;
+
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.client.sync.SyncConfigNodeIServiceClient;
+import org.apache.iotdb.commons.pipe.agent.plugin.builtin.BuiltinPipePlugin;
+import org.apache.iotdb.confignode.rpc.thrift.TCreatePipeReq;
+import org.apache.iotdb.consensus.ConsensusFactory;
+import org.apache.iotdb.it.env.MultiEnvFactory;
+import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper;
+import org.apache.iotdb.it.framework.IoTDBTestRunner;
+import org.apache.iotdb.itbase.category.MultiClusterIT2TableModel;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/** Test pipe's basic functionalities under multiple cluster and consensus protocol settings. */
+@RunWith(IoTDBTestRunner.class)
+@Category({MultiClusterIT2TableModel.class})
+public class IoTDBPipeProtocolIT extends AbstractPipeTableModelTestIT {
+ @Override
+ @Before
+ public void setUp() {
+ MultiEnvFactory.createEnv(2);
+ senderEnv = MultiEnvFactory.getEnv(0);
+ receiverEnv = MultiEnvFactory.getEnv(1);
+ }
+
+ private void innerSetUp(
+ final String configNodeConsensus,
+ final String schemaRegionConsensus,
+ final String dataRegionConsensus,
+ final int configNodesNum,
+ final int dataNodesNum,
+ int schemaRegionReplicationFactor,
+ int dataRegionReplicationFactor) {
+ schemaRegionReplicationFactor = Math.min(schemaRegionReplicationFactor, dataNodesNum);
+ dataRegionReplicationFactor = Math.min(dataRegionReplicationFactor, dataNodesNum);
+
+ // TODO: delete ratis configurations
+ senderEnv
+ .getConfig()
+ .getCommonConfig()
+ .setAutoCreateSchemaEnabled(true)
+ .setConfigNodeConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS)
+ .setSchemaRegionConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS)
+ .setDataRegionConsensusProtocolClass(dataRegionConsensus)
+ .setSchemaReplicationFactor(schemaRegionReplicationFactor)
+ .setDataReplicationFactor(dataRegionReplicationFactor);
+ receiverEnv
+ .getConfig()
+ .getCommonConfig()
+ .setAutoCreateSchemaEnabled(true)
+ .setConfigNodeConsensusProtocolClass(configNodeConsensus)
+ .setSchemaRegionConsensusProtocolClass(schemaRegionConsensus)
+ .setDataRegionConsensusProtocolClass(dataRegionConsensus)
+ .setSchemaReplicationFactor(schemaRegionReplicationFactor)
+ .setDataReplicationFactor(dataRegionReplicationFactor);
+
+ // 10 min, assert that the operations will not time out
+ senderEnv.getConfig().getCommonConfig().setCnConnectionTimeoutMs(600000);
+ receiverEnv.getConfig().getCommonConfig().setCnConnectionTimeoutMs(600000);
+
+ senderEnv.initClusterEnvironment(configNodesNum, dataNodesNum);
+ receiverEnv.initClusterEnvironment(configNodesNum, dataNodesNum);
+ }
+
+ @Test
+ public void test1C1DWithRatisRatisIot() throws Exception {
+ innerSetUp(
+ ConsensusFactory.RATIS_CONSENSUS,
+ ConsensusFactory.RATIS_CONSENSUS,
+ ConsensusFactory.IOT_CONSENSUS,
+ 1,
+ 1,
+ 1,
+ 1);
+ doTest();
+ }
+
+ @Test
+ public void test1C1DWithSimpleSimpleIot() throws Exception {
+ innerSetUp(
+ ConsensusFactory.SIMPLE_CONSENSUS,
+ ConsensusFactory.SIMPLE_CONSENSUS,
+ ConsensusFactory.IOT_CONSENSUS,
+ 1,
+ 1,
+ 1,
+ 1);
+ doTest();
+ }
+
+ @Test
+ public void test1C1DWithRatisRatisSimple() throws Exception {
+ innerSetUp(
+ ConsensusFactory.RATIS_CONSENSUS,
+ ConsensusFactory.RATIS_CONSENSUS,
+ ConsensusFactory.SIMPLE_CONSENSUS,
+ 1,
+ 1,
+ 1,
+ 1);
+ doTest();
+ }
+
+ @Test
+ public void test3C3DWith3SchemaRegionFactor3DataRegionFactor() throws Exception {
+ innerSetUp(
+ ConsensusFactory.RATIS_CONSENSUS,
+ ConsensusFactory.RATIS_CONSENSUS,
+ ConsensusFactory.IOT_CONSENSUS,
+ 3,
+ 3,
+ 3,
+ 3);
+ doTest();
+ }
+
+ @Test
+ public void test3C3DWith3SchemaRegionFactor2DataRegionFactor() throws Exception {
+ innerSetUp(
+ ConsensusFactory.RATIS_CONSENSUS,
+ ConsensusFactory.RATIS_CONSENSUS,
+ ConsensusFactory.IOT_CONSENSUS,
+ 3,
+ 3,
+ 3,
+ 2);
+ doTest();
+ }
+
+ @Test
+ public void testPipeOnBothSenderAndReceiver() throws Exception {
+ senderEnv
+ .getConfig()
+ .getCommonConfig()
+ .setAutoCreateSchemaEnabled(true)
+ .setConfigNodeConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS)
+ .setSchemaRegionConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS)
+ .setDataRegionConsensusProtocolClass(ConsensusFactory.IOT_CONSENSUS)
+ .setSchemaReplicationFactor(3)
+ .setDataReplicationFactor(2);
+ receiverEnv
+ .getConfig()
+ .getCommonConfig()
+ .setAutoCreateSchemaEnabled(true)
+ .setConfigNodeConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS)
+ .setSchemaRegionConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS)
+ .setDataRegionConsensusProtocolClass(ConsensusFactory.IOT_CONSENSUS)
+ .setSchemaReplicationFactor(1)
+ .setDataReplicationFactor(1);
+
+ // 10 min, assert that the operations will not time out
+ senderEnv.getConfig().getCommonConfig().setCnConnectionTimeoutMs(600000);
+ receiverEnv.getConfig().getCommonConfig().setCnConnectionTimeoutMs(600000);
+
+ senderEnv.initClusterEnvironment(3, 3);
+ receiverEnv.initClusterEnvironment(1, 1);
+
+ final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+ final String receiverIp = receiverDataNode.getIp();
+ final int receiverPort = receiverDataNode.getPort();
+ boolean insertResult = true;
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+
+ TableModelUtils.createDataBaseAndTable(senderEnv, "test", "test");
+ insertResult = TableModelUtils.insertData("test", "test", 0, 100, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ final Map<String, String> extractorAttributes = new HashMap<>();
+ final Map<String, String> processorAttributes = new HashMap<>();
+ final Map<String, String> connectorAttributes = new HashMap<>();
+
+ extractorAttributes.put("capture.table", "true");
+ extractorAttributes.put("database-name", "test");
+ extractorAttributes.put("table-name", "test.*");
+ extractorAttributes.put("inclusion", "data.insert");
+ extractorAttributes.put("mode.streaming", "true");
+ extractorAttributes.put("mode.snapshot", "false");
+ extractorAttributes.put("mode.strict", "true");
+
+ connectorAttributes.put("connector", "iotdb-thrift-connector");
+ connectorAttributes.put("connector.batch.enable", "false");
+ connectorAttributes.put("connector.ip", receiverIp);
+ connectorAttributes.put("connector.port", Integer.toString(receiverPort));
+
+ final TSStatus status =
+ client.createPipe(
+ new TCreatePipeReq("p1", connectorAttributes)
+ .setExtractorAttributes(extractorAttributes)
+ .setProcessorAttributes(processorAttributes));
+
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("p1").getCode());
+
+ TableModelUtils.assertCountData("test", "test", 100, receiverEnv);
+
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.stopPipe("p1").getCode());
+ }
+
+ final DataNodeWrapper senderDataNode = senderEnv.getDataNodeWrapper(0);
+ final String senderIp = senderDataNode.getIp();
+ final int senderPort = senderDataNode.getPort();
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) receiverEnv.getLeaderConfigNodeConnection()) {
+
+ TableModelUtils.createDataBaseAndTable(receiverEnv, "test1", "test1");
+ insertResult = TableModelUtils.insertData("test1", "test1", 0, 100, receiverEnv);
+ if (!insertResult) {
+ return;
+ }
+ final Map<String, String> extractorAttributes = new HashMap<>();
+ final Map<String, String> processorAttributes = new HashMap<>();
+ final Map<String, String> connectorAttributes = new HashMap<>();
+
+ extractorAttributes.put("capture.table", "true");
+ extractorAttributes.put("database-name", "test.*");
+ extractorAttributes.put("table-name", "test.*");
+ extractorAttributes.put("inclusion", "data.insert");
+ extractorAttributes.put("mode.streaming", "true");
+ extractorAttributes.put("mode.snapshot", "false");
+ extractorAttributes.put("mode.strict", "true");
+
+ connectorAttributes.put("connector", "iotdb-thrift-connector");
+ connectorAttributes.put("connector.batch.enable", "false");
+ connectorAttributes.put("connector.ip", senderIp);
+ connectorAttributes.put("connector.port", Integer.toString(senderPort));
+
+ final TSStatus status =
+ client.createPipe(
+ new TCreatePipeReq("p1", connectorAttributes)
+ .setExtractorAttributes(extractorAttributes)
+ .setProcessorAttributes(processorAttributes));
+
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("p1").getCode());
+
+ TableModelUtils.assertCountData("test1", "test1", 100, senderEnv);
+ }
+ }
+
+ private void doTest() throws Exception {
+ final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+
+ final String receiverIp = receiverDataNode.getIp();
+ final int receiverPort = receiverDataNode.getPort();
+ boolean insertResult = true;
+
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+
+ TableModelUtils.createDataBaseAndTable(senderEnv, "test", "test");
+ insertResult = TableModelUtils.insertData("test", "test", 0, 100, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ final Map<String, String> extractorAttributes = new HashMap<>();
+ final Map<String, String> processorAttributes = new HashMap<>();
+ final Map<String, String> connectorAttributes = new HashMap<>();
+
+ extractorAttributes.put("capture.table", "true");
+ extractorAttributes.put("database-name", "test.*");
+ extractorAttributes.put("table-name", "test.*");
+ extractorAttributes.put("inclusion", "data.insert");
+ extractorAttributes.put("mode.streaming", "true");
+ extractorAttributes.put("mode.snapshot", "false");
+ extractorAttributes.put("mode.strict", "true");
+
+ connectorAttributes.put("connector", "iotdb-thrift-connector");
+ connectorAttributes.put("connector.batch.enable", "false");
+ connectorAttributes.put("connector.ip", receiverIp);
+ connectorAttributes.put("connector.port", Integer.toString(receiverPort));
+
+ final TSStatus status =
+ client.createPipe(
+ new TCreatePipeReq("p1", connectorAttributes)
+ .setExtractorAttributes(extractorAttributes)
+ .setProcessorAttributes(processorAttributes));
+
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("p1").getCode());
+
+ TableModelUtils.assertData("test", "test", 0, 100, receiverEnv);
+
+ insertResult = TableModelUtils.insertData("test", "test", 100, 200, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ TableModelUtils.assertData("test", "test", 0, 200, receiverEnv);
+
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.stopPipe("p1").getCode());
+
+ insertResult = TableModelUtils.insertData("test", "test", 200, 300, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ TableModelUtils.assertData("test", "test", 0, 200, receiverEnv);
+ }
+ }
+
+ @Test
+ public void testSyncConnectorUseNodeUrls() throws Exception {
+ doTestUseNodeUrls(BuiltinPipePlugin.IOTDB_THRIFT_SYNC_CONNECTOR.getPipePluginName());
+ }
+
+ @Test
+ public void testAsyncConnectorUseNodeUrls() throws Exception {
+ doTestUseNodeUrls(BuiltinPipePlugin.IOTDB_THRIFT_ASYNC_CONNECTOR.getPipePluginName());
+ }
+
+ @Test
+ public void testAirGapConnectorUseNodeUrls() throws Exception {
+ doTestUseNodeUrls(BuiltinPipePlugin.IOTDB_AIR_GAP_CONNECTOR.getPipePluginName());
+ }
+
+ private void doTestUseNodeUrls(String connectorName) throws Exception {
+ senderEnv
+ .getConfig()
+ .getCommonConfig()
+ .setAutoCreateSchemaEnabled(true)
+ .setPipeAirGapReceiverEnabled(true)
+ .setConfigNodeConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS)
+ .setSchemaRegionConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS)
+ .setDataRegionConsensusProtocolClass(ConsensusFactory.IOT_CONSENSUS)
+ .setSchemaReplicationFactor(1)
+ .setDataReplicationFactor(1)
+ .setEnableSeqSpaceCompaction(false)
+ .setEnableUnseqSpaceCompaction(false)
+ .setEnableCrossSpaceCompaction(false);
+ receiverEnv
+ .getConfig()
+ .getCommonConfig()
+ .setAutoCreateSchemaEnabled(true)
+ .setPipeAirGapReceiverEnabled(true)
+ .setConfigNodeConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS)
+ .setSchemaRegionConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS)
+ .setDataRegionConsensusProtocolClass(ConsensusFactory.IOT_CONSENSUS)
+ .setSchemaReplicationFactor(3)
+ .setDataReplicationFactor(2);
+
+ // 10 min, assert that the operations will not time out
+ senderEnv.getConfig().getCommonConfig().setCnConnectionTimeoutMs(600000);
+ receiverEnv.getConfig().getCommonConfig().setCnConnectionTimeoutMs(600000);
+
+ senderEnv.initClusterEnvironment(1, 1);
+ receiverEnv.initClusterEnvironment(1, 3);
+
+ final StringBuilder nodeUrlsBuilder = new StringBuilder();
+ boolean insertResult = true;
+ for (final DataNodeWrapper wrapper : receiverEnv.getDataNodeWrapperList()) {
+ if (connectorName.equals(BuiltinPipePlugin.IOTDB_AIR_GAP_CONNECTOR.getPipePluginName())) {
+ // Use default port for convenience
+ nodeUrlsBuilder
+ .append(wrapper.getIp())
+ .append(":")
+ .append(wrapper.getPipeAirGapReceiverPort())
+ .append(",");
+ } else {
+ nodeUrlsBuilder.append(wrapper.getIpAndPortString()).append(",");
+ }
+ }
+
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+
+ TableModelUtils.createDataBaseAndTable(senderEnv, "test", "test");
+ insertResult = TableModelUtils.insertData("test", "test", 0, 100, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ final Map<String, String> extractorAttributes = new HashMap<>();
+ final Map<String, String> processorAttributes = new HashMap<>();
+ final Map<String, String> connectorAttributes = new HashMap<>();
+
+ connectorAttributes.put("connector", connectorName);
+ connectorAttributes.put("connector.batch.enable", "false");
+ connectorAttributes.put("connector.batch.max-delay-seconds", "1");
+ connectorAttributes.put("connector.batch.size-bytes", "2048");
+ connectorAttributes.put("connector.node-urls", nodeUrlsBuilder.toString());
+
+ extractorAttributes.put("capture.table", "true");
+ extractorAttributes.put("database-name", "test.*");
+ extractorAttributes.put("table-name", "test.*");
+ extractorAttributes.put("inclusion", "data.insert");
+ extractorAttributes.put("mode.streaming", "true");
+ extractorAttributes.put("mode.snapshot", "false");
+ extractorAttributes.put("mode.strict", "true");
+
+ // Test forced-log mode, in open releases this might be "file"
+ extractorAttributes.put("realtime.mode", "forced-log");
+
+ TSStatus status =
+ client.createPipe(
+ new TCreatePipeReq("p1", connectorAttributes)
+ .setExtractorAttributes(extractorAttributes)
+ .setProcessorAttributes(processorAttributes));
+
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("p1").getCode());
+
+ insertResult = TableModelUtils.insertData("test", "test", 100, 200, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ TableModelUtils.assertData("test", "test", 0, 200, receiverEnv);
+
+ insertResult = TableModelUtils.insertData("test", "test", 200, 300, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ TableModelUtils.assertData("test", "test", 0, 300, receiverEnv);
+
+ extractorAttributes.replace("realtime.mode", "file");
+
+ status =
+ client.createPipe(
+ new TCreatePipeReq("p2", connectorAttributes)
+ .setExtractorAttributes(extractorAttributes)
+ .setProcessorAttributes(processorAttributes));
+
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("p2").getCode());
+
+ insertResult = TableModelUtils.insertData("test", "test", 300, 400, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ TableModelUtils.assertData("test", "test", 0, 400, receiverEnv);
+ }
+ }
+}
diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/tablemodel/IoTDBPipeSwitchStatusIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/tablemodel/IoTDBPipeSwitchStatusIT.java
new file mode 100644
index 00000000000..1420b3cec5d
--- /dev/null
+++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/tablemodel/IoTDBPipeSwitchStatusIT.java
@@ -0,0 +1,348 @@
+/*
+ * 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.pipe.it.tablemodel;
+
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.client.sync.SyncConfigNodeIServiceClient;
+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.it.env.cluster.node.DataNodeWrapper;
+import org.apache.iotdb.it.framework.IoTDBTestRunner;
+import org.apache.iotdb.itbase.category.MultiClusterIT2TableModel;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+@RunWith(IoTDBTestRunner.class)
+@Category({MultiClusterIT2TableModel.class})
+public class IoTDBPipeSwitchStatusIT extends AbstractPipeTableModelTestIT {
+ @Test
+ public void testPipeSwitchStatus() throws Exception {
+ final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+
+ final String receiverIp = receiverDataNode.getIp();
+ final int receiverPort = receiverDataNode.getPort();
+
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+ final Map<String, String> extractorAttributes = new HashMap<>();
+ final Map<String, String> processorAttributes = new HashMap<>();
+ final Map<String, String> connectorAttributes = new HashMap<>();
+
+ extractorAttributes.put("capture.table", "true");
+ extractorAttributes.put("database-name", "test");
+ extractorAttributes.put("table-name", "test.*");
+ extractorAttributes.put("inclusion", "data.insert");
+ extractorAttributes.put("mode.streaming", "true");
+ extractorAttributes.put("mode.snapshot", "false");
+ extractorAttributes.put("mode.strict", "true");
+ extractorAttributes.put("start-time", "1");
+ extractorAttributes.put("end-time", "2");
+
+ connectorAttributes.put("connector", "iotdb-thrift-connector");
+ connectorAttributes.put("connector.batch.enable", "false");
+ connectorAttributes.put("connector.ip", receiverIp);
+ connectorAttributes.put("connector.port", Integer.toString(receiverPort));
+
+ TSStatus status =
+ client.createPipe(
+ new TCreatePipeReq("p1", connectorAttributes)
+ .setExtractorAttributes(extractorAttributes)
+ .setProcessorAttributes(processorAttributes));
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+
+ status =
+ client.createPipe(
+ new TCreatePipeReq("p2", connectorAttributes)
+ .setExtractorAttributes(extractorAttributes)
+ .setProcessorAttributes(processorAttributes));
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+
+ status =
+ client.createPipe(
+ new TCreatePipeReq("p3", connectorAttributes)
+ .setExtractorAttributes(extractorAttributes)
+ .setProcessorAttributes(processorAttributes));
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+
+ List<TShowPipeInfo> showPipeResult = client.showPipe(new TShowPipeReq()).pipeInfoList;
+ Assert.assertTrue(
+ showPipeResult.stream().anyMatch((o) -> o.id.equals("p1") && o.state.equals("RUNNING")));
+ Assert.assertTrue(
+ showPipeResult.stream().anyMatch((o) -> o.id.equals("p2") && o.state.equals("RUNNING")));
+ Assert.assertTrue(
+ showPipeResult.stream().anyMatch((o) -> o.id.equals("p3") && o.state.equals("RUNNING")));
+
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("p1").getCode());
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("p2").getCode());
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.dropPipe("p3").getCode());
+
+ showPipeResult = client.showPipe(new TShowPipeReq()).pipeInfoList;
+ Assert.assertTrue(
+ showPipeResult.stream().anyMatch((o) -> o.id.equals("p1") && o.state.equals("RUNNING")));
+ Assert.assertTrue(
+ showPipeResult.stream().anyMatch((o) -> o.id.equals("p2") && o.state.equals("RUNNING")));
+ Assert.assertFalse(showPipeResult.stream().anyMatch((o) -> o.id.equals("p3")));
+
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.stopPipe("p1").getCode());
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.dropPipe("p2").getCode());
+
+ showPipeResult = client.showPipe(new TShowPipeReq()).pipeInfoList;
+ Assert.assertTrue(
+ showPipeResult.stream().anyMatch((o) -> o.id.equals("p1") && o.state.equals("STOPPED")));
+ Assert.assertFalse(showPipeResult.stream().anyMatch((o) -> o.id.equals("p2")));
+
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.dropPipe("p1").getCode());
+
+ showPipeResult = client.showPipe(new TShowPipeReq()).pipeInfoList;
+ Assert.assertFalse(showPipeResult.stream().anyMatch((o) -> o.id.equals("p1")));
+ }
+ }
+
+ @Test
+ public void testPipeIllegallySwitchStatus() throws Exception {
+ final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+
+ final String receiverIp = receiverDataNode.getIp();
+ final int receiverPort = receiverDataNode.getPort();
+
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+ final Map<String, String> extractorAttributes = new HashMap<>();
+ final Map<String, String> processorAttributes = new HashMap<>();
+ final Map<String, String> connectorAttributes = new HashMap<>();
+
+ extractorAttributes.put("capture.table", "true");
+ extractorAttributes.put("database-name", "test");
+ extractorAttributes.put("table-name", "test.*");
+ extractorAttributes.put("inclusion", "data.insert");
+ extractorAttributes.put("mode.streaming", "true");
+ extractorAttributes.put("mode.snapshot", "false");
+ extractorAttributes.put("mode.strict", "true");
+ extractorAttributes.put("start-time", "1");
+ extractorAttributes.put("end-time", "2");
+
+ connectorAttributes.put("connector", "iotdb-thrift-connector");
+ connectorAttributes.put("connector.batch.enable", "false");
+ connectorAttributes.put("connector.ip", receiverIp);
+ connectorAttributes.put("connector.port", Integer.toString(receiverPort));
+
+ TSStatus status =
+ client.createPipe(
+ new TCreatePipeReq("p1", connectorAttributes)
+ .setExtractorAttributes(extractorAttributes)
+ .setProcessorAttributes(processorAttributes));
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+
+ List<TShowPipeInfo> showPipeResult = client.showPipe(new TShowPipeReq()).pipeInfoList;
+ Assert.assertTrue(
+ showPipeResult.stream().anyMatch((o) -> o.id.equals("p1") && o.state.equals("RUNNING")));
+
+ status =
+ client.createPipe(
+ new TCreatePipeReq("p1", connectorAttributes)
+ .setExtractorAttributes(extractorAttributes)
+ .setProcessorAttributes(processorAttributes));
+ Assert.assertEquals(TSStatusCode.PIPE_ERROR.getStatusCode(), status.getCode());
+
+ showPipeResult = client.showPipe(new TShowPipeReq()).pipeInfoList;
+ Assert.assertEquals(1, showPipeResult.stream().filter((o) -> o.id.equals("p1")).count());
+
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("p1").getCode());
+ showPipeResult = client.showPipe(new TShowPipeReq()).pipeInfoList;
+ Assert.assertTrue(
+ showPipeResult.stream().anyMatch((o) -> o.id.equals("p1") && o.state.equals("RUNNING")));
+
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.stopPipe("p1").getCode());
+ showPipeResult = client.showPipe(new TShowPipeReq()).pipeInfoList;
+ Assert.assertTrue(
+ showPipeResult.stream().anyMatch((o) -> o.id.equals("p1") && o.state.equals("STOPPED")));
+ Assert.assertEquals(1, showPipeResult.stream().filter((o) -> o.id.equals("p1")).count());
+
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.stopPipe("p1").getCode());
+ showPipeResult = client.showPipe(new TShowPipeReq()).pipeInfoList;
+ Assert.assertTrue(
+ showPipeResult.stream().anyMatch((o) -> o.id.equals("p1") && o.state.equals("STOPPED")));
+ Assert.assertEquals(1, showPipeResult.stream().filter((o) -> o.id.equals("p1")).count());
+ }
+ }
+
+ @Test
+ public void testDropPipeAndCreateAgain() throws Exception {
+ final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+
+ final String receiverIp = receiverDataNode.getIp();
+ final int receiverPort = receiverDataNode.getPort();
+ boolean insertResult = true;
+
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+
+ TableModelUtils.createDataBaseAndTable(senderEnv, "test", "test");
+ insertResult = TableModelUtils.insertData("test", "test", 0, 100, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ final Map<String, String> extractorAttributes = new HashMap<>();
+ final Map<String, String> processorAttributes = new HashMap<>();
+ final Map<String, String> connectorAttributes = new HashMap<>();
+
+ extractorAttributes.put("capture.table", "true");
+ extractorAttributes.put("database-name", "test");
+ extractorAttributes.put("table-name", "test.*");
+ extractorAttributes.put("inclusion", "data.insert");
+ extractorAttributes.put("mode.streaming", "true");
+ extractorAttributes.put("mode.snapshot", "false");
+ extractorAttributes.put("mode.strict", "true");
+ extractorAttributes.put("start-time", "0");
+ extractorAttributes.put("end-time", "200");
+
+ connectorAttributes.put("connector", "iotdb-thrift-connector");
+ connectorAttributes.put("connector.batch.enable", "false");
+ connectorAttributes.put("connector.ip", receiverIp);
+ connectorAttributes.put("connector.port", Integer.toString(receiverPort));
+
+ TSStatus status =
+ client.createPipe(
+ new TCreatePipeReq("p1", connectorAttributes)
+ .setExtractorAttributes(extractorAttributes)
+ .setProcessorAttributes(processorAttributes));
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("p1").getCode());
+ List<TShowPipeInfo> showPipeResult = client.showPipe(new TShowPipeReq()).pipeInfoList;
+ Assert.assertTrue(
+ showPipeResult.stream().anyMatch((o) -> o.id.equals("p1") && o.state.equals("RUNNING")));
+
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.dropPipe("p1").getCode());
+ showPipeResult = client.showPipe(new TShowPipeReq()).pipeInfoList;
+ Assert.assertEquals(0, showPipeResult.size());
+
+ status =
+ client.createPipe(
+ new TCreatePipeReq("p1", connectorAttributes)
+ .setExtractorAttributes(extractorAttributes)
+ .setProcessorAttributes(processorAttributes));
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+ showPipeResult = client.showPipe(new TShowPipeReq()).pipeInfoList;
+ Assert.assertEquals(1, showPipeResult.stream().filter((o) -> o.id.equals("p1")).count());
+ }
+ }
+
+ @Test
+ public void testWrongPipeName() throws Exception {
+ final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+
+ final String receiverIp = receiverDataNode.getIp();
+ final int receiverPort = receiverDataNode.getPort();
+
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+ final Map<String, String> extractorAttributes = new HashMap<>();
+ final Map<String, String> processorAttributes = new HashMap<>();
+ final Map<String, String> connectorAttributes = new HashMap<>();
+
+ extractorAttributes.put("capture.table", "true");
+ extractorAttributes.put("database-name", "test");
+ extractorAttributes.put("table-name", "test.*");
+ extractorAttributes.put("inclusion", "data.insert");
+ extractorAttributes.put("mode.streaming", "true");
+ extractorAttributes.put("mode.snapshot", "false");
+ extractorAttributes.put("mode.strict", "true");
+ extractorAttributes.put("start-time", "0");
+ extractorAttributes.put("end-time", "200");
+
+ connectorAttributes.put("connector", "iotdb-thrift-connector");
+ connectorAttributes.put("connector.batch.enable", "false");
+ connectorAttributes.put("connector.ip", receiverIp);
+ connectorAttributes.put("connector.port", Integer.toString(receiverPort));
+
+ final TSStatus status =
+ client.createPipe(
+ new TCreatePipeReq("p1", connectorAttributes)
+ .setExtractorAttributes(extractorAttributes)
+ .setProcessorAttributes(processorAttributes));
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+ Assert.assertEquals(TSStatusCode.PIPE_ERROR.getStatusCode(), client.startPipe("").getCode());
+ Assert.assertEquals(
+ TSStatusCode.PIPE_ERROR.getStatusCode(), client.startPipe("p0").getCode());
+ Assert.assertEquals(TSStatusCode.PIPE_ERROR.getStatusCode(), client.startPipe("p").getCode());
+ Assert.assertEquals(TSStatusCode.PIPE_ERROR.getStatusCode(), client.startPipe("*").getCode());
+ List<TShowPipeInfo> showPipeResult = client.showPipe(new TShowPipeReq()).pipeInfoList;
+ Assert.assertTrue(
+ showPipeResult.stream().anyMatch((o) -> o.id.equals("p1") && o.state.equals("RUNNING")));
+
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("p1").getCode());
+ showPipeResult = client.showPipe(new TShowPipeReq()).pipeInfoList;
+ Assert.assertTrue(
+ showPipeResult.stream().anyMatch((o) -> o.id.equals("p1") && o.state.equals("RUNNING")));
+
+ Assert.assertEquals(TSStatusCode.PIPE_ERROR.getStatusCode(), client.stopPipe("").getCode());
+ Assert.assertEquals(TSStatusCode.PIPE_ERROR.getStatusCode(), client.stopPipe("p0").getCode());
+ Assert.assertEquals(TSStatusCode.PIPE_ERROR.getStatusCode(), client.stopPipe("p").getCode());
+ Assert.assertEquals(TSStatusCode.PIPE_ERROR.getStatusCode(), client.stopPipe("*").getCode());
+ showPipeResult = client.showPipe(new TShowPipeReq()).pipeInfoList;
+ Assert.assertTrue(
+ showPipeResult.stream().anyMatch((o) -> o.id.equals("p1") && o.state.equals("RUNNING")));
+
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.stopPipe("p1").getCode());
+ showPipeResult = client.showPipe(new TShowPipeReq()).pipeInfoList;
+ Assert.assertTrue(
+ showPipeResult.stream().anyMatch((o) -> o.id.equals("p1") && o.state.equals("STOPPED")));
+
+ Assert.assertEquals(
+ TSStatusCode.PIPE_NOT_EXIST_ERROR.getStatusCode(), client.dropPipe("").getCode());
+ Assert.assertEquals(
+ TSStatusCode.PIPE_NOT_EXIST_ERROR.getStatusCode(), client.dropPipe("p0").getCode());
+ Assert.assertEquals(
+ TSStatusCode.PIPE_NOT_EXIST_ERROR.getStatusCode(), client.dropPipe("p").getCode());
+ Assert.assertEquals(
+ TSStatusCode.PIPE_NOT_EXIST_ERROR.getStatusCode(), client.dropPipe("*").getCode());
+ showPipeResult = client.showPipe(new TShowPipeReq()).pipeInfoList;
+ Assert.assertTrue(
+ showPipeResult.stream().anyMatch((o) -> o.id.equals("p1") && o.state.equals("STOPPED")));
+
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.dropPipe("p1").getCode());
+ showPipeResult = client.showPipe(new TShowPipeReq()).pipeInfoList;
+ Assert.assertFalse(showPipeResult.stream().anyMatch((o) -> o.id.equals("p1")));
+ }
+ }
+}
diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/tablemodel/IoTDBPipeSyntaxIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/tablemodel/IoTDBPipeSyntaxIT.java
new file mode 100644
index 00000000000..24b1c92563b
--- /dev/null
+++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/tablemodel/IoTDBPipeSyntaxIT.java
@@ -0,0 +1,685 @@
+/*
+ * 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.pipe.it.tablemodel;
+
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.client.sync.SyncConfigNodeIServiceClient;
+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.it.env.cluster.node.DataNodeWrapper;
+import org.apache.iotdb.it.framework.IoTDBTestRunner;
+import org.apache.iotdb.itbase.category.MultiClusterIT2TableModel;
+import org.apache.iotdb.itbase.env.BaseEnv;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import java.sql.Connection;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.Assert.fail;
+
+@RunWith(IoTDBTestRunner.class)
+@Category({MultiClusterIT2TableModel.class})
+public class IoTDBPipeSyntaxIT extends AbstractPipeTableModelTestIT {
+ @Test
+ public void testValidPipeName() throws Exception {
+ final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+
+ final String receiverIp = receiverDataNode.getIp();
+ final int receiverPort = receiverDataNode.getPort();
+
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+
+ final List<String> validPipeNames =
+ Arrays.asList("Pipe_1", "\"`33`\"", "root", "\"`ä¸æ–‡`\"", "\"`with`\"");
+ final List<String> expectedPipeNames =
+ Arrays.asList("Pipe_1", "`33`", "root", "`ä¸æ–‡`", "`with`");
+ for (final String pipeName : validPipeNames) {
+ try (final Connection connection = senderEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ final Statement statement = connection.createStatement()) {
+ statement.execute(
+ String.format(
+ "create pipe %s"
+ + " with source ( "
+ + "'capture.table'='true',"
+ + "'database-name'='test',"
+ + "'table-name'='test',"
+ + "'mode.streaming'='true',"
+ + "'mode.strict'='true',"
+ + "'mode.snapshot'='false',"
+ + "'start-time'='1',"
+ + "'end-time'='2')"
+ + " with connector ("
+ + "'connector'='iotdb-thrift-connector',"
+ + "'connector.ip'='%s',"
+ + "'connector.port'='%s',"
+ + "'connector.batch.enable'='false')",
+ pipeName, receiverIp, receiverPort));
+ } catch (SQLException e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ List<TShowPipeInfo> showPipeResult = client.showPipe(new TShowPipeReq()).pipeInfoList;
+ for (final String pipeName : expectedPipeNames) {
+ Assert.assertTrue(
+ showPipeResult.stream()
+ .anyMatch((o) -> o.id.equals(pipeName) && o.state.equals("RUNNING")));
+ }
+
+ for (final String pipeName : validPipeNames) {
+ try (final Connection connection = senderEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ final Statement statement = connection.createStatement()) {
+ statement.execute(String.format("drop pipe %s", pipeName));
+ } catch (SQLException e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ showPipeResult = client.showPipe(new TShowPipeReq()).pipeInfoList;
+ Assert.assertEquals(0, showPipeResult.size());
+ }
+ }
+
+ @Test
+ public void testRevertParameterOrder() {
+ final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+
+ final String receiverIp = receiverDataNode.getIp();
+ final int receiverPort = receiverDataNode.getPort();
+
+ try (final Connection connection = senderEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ final Statement statement = connection.createStatement()) {
+ statement.execute(
+ String.format(
+ "create pipe p1"
+ + " with source ( "
+ + "'capture.table'='test',"
+ + "'database-name'='test',"
+ + "'table-name'='test',"
+ + "'mode.streaming'='true',"
+ + "'mode.strict'='true',"
+ + "'mode.snapshot'='false',"
+ + "'start-time'='1',"
+ + "'end-time'='2')"
+ + " with connector ("
+ + "'connector.batch.enable'='false', "
+ + "'connector.port'='%s',"
+ + "'connector.ip'='%s',"
+ + "'connector'='iotdb-thrift-connector')",
+ receiverIp, receiverPort));
+ fail();
+ } catch (SQLException ignore) {
+ }
+ }
+
+ @Test
+ public void testRevertStageOrder() throws Exception {
+ final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+
+ final String receiverIp = receiverDataNode.getIp();
+ final int receiverPort = receiverDataNode.getPort();
+
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+
+ try (final Connection connection = senderEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ final Statement statement = connection.createStatement()) {
+ statement.execute(
+ String.format(
+ "create pipe p1"
+ + " with connector ("
+ + "'connector.batch.enable'='false', "
+ + "'connector.port'='%s',"
+ + "'connector.ip'='%s',"
+ + "'connector'='iotdb-thrift-connector') "
+ + " with source ( "
+ + "'capture.table'='true',"
+ + "'database-name'='test',"
+ + "'table-name'='test',"
+ + "'mode.streaming'='true',"
+ + "'mode.strict'='true',"
+ + "'mode.snapshot'='false',"
+ + "'start-time'='1',"
+ + "'end-time'='2')",
+ receiverIp, receiverPort));
+ fail();
+ } catch (SQLException ignored) {
+ }
+
+ final List<TShowPipeInfo> showPipeResult = client.showPipe(new TShowPipeReq()).pipeInfoList;
+ Assert.assertEquals(0, showPipeResult.size());
+ }
+ }
+
+ @Test
+ public void testMissingStage() throws Exception {
+ final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+
+ final String receiverIp = receiverDataNode.getIp();
+ final int receiverPort = receiverDataNode.getPort();
+
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+ try (final Connection connection = senderEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ final Statement statement = connection.createStatement()) {
+ statement.execute("create pipe p1");
+ fail();
+ } catch (SQLException ignored) {
+ }
+
+ try (final Connection connection = senderEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ final Statement statement = connection.createStatement()) {
+ statement.execute("create pipe p2 with extractor ('extractor'='iotdb-extractor')");
+ fail();
+ } catch (SQLException ignored) {
+ }
+
+ try (final Connection connection = senderEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ final Statement statement = connection.createStatement()) {
+ statement.execute(
+ "create pipe p3"
+ + " with extractor ('extractor'='iotdb-extractor')"
+ + " with processor ('processor'='do-nothing-processor')");
+ fail();
+ } catch (SQLException ignored) {
+ }
+
+ try (final Connection connection = senderEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ final Statement statement = connection.createStatement()) {
+ statement.execute(
+ String.format(
+ "create pipe p4"
+ + " with connector ("
+ + "'connector'='iotdb-thrift-connector',"
+ + "'connector.ip'='%s',"
+ + "'connector.port'='%s',"
+ + "'connector.batch.enable'='false')",
+ receiverIp, receiverPort));
+ } catch (SQLException e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+
+ try (final Connection connection = senderEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ final Statement statement = connection.createStatement()) {
+ statement.execute(
+ String.format(
+ "create pipe p5"
+ + " with extractor ('extractor'='iotdb-extractor')"
+ + " with processor ('processor'='do-nothing-processor')"
+ + " with connector ("
+ + "'connector'='iotdb-thrift-connector',"
+ + "'connector.ip'='%s',"
+ + "'connector.port'='%s',"
+ + "'connector.batch.enable'='false')",
+ receiverIp, receiverPort));
+ } catch (SQLException e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+
+ final List<TShowPipeInfo> showPipeResult = client.showPipe(new TShowPipeReq()).pipeInfoList;
+ Assert.assertEquals(2, showPipeResult.size());
+ }
+ }
+
+ @Test
+ public void testInvalidParameter() throws Exception {
+ final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+
+ final String receiverIp = receiverDataNode.getIp();
+ final int receiverPort = receiverDataNode.getPort();
+
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+ try (final Connection connection = senderEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ final Statement statement = connection.createStatement()) {
+ statement.execute(
+ String.format(
+ "create pipe p1"
+ + " with extractor ()"
+ + " with processor ()"
+ + " with connector ("
+ + "'connector'='iotdb-thrift-connector',"
+ + "'connector.ip'='%s',"
+ + "'connector.port'='%s',"
+ + "'connector.batch.enable'='false')",
+ receiverIp, receiverPort));
+ } catch (SQLException e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+
+ try (final Connection connection = senderEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ final Statement statement = connection.createStatement()) {
+ statement.execute(
+ String.format(
+ "create pipe p3"
+ + " with extractor ()"
+ + " with processor ('processor'='invalid-param')"
+ + " with connector ("
+ + "'connector'='iotdb-thrift-connector',"
+ + "'connector.ip'='%s',"
+ + "'connector.port'='%s',"
+ + "'connector.batch.enable'='false')",
+ receiverIp, receiverPort));
+ fail();
+ } catch (SQLException ignored) {
+ }
+
+ try (final Connection connection = senderEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ final Statement statement = connection.createStatement()) {
+ statement.execute(
+ String.format(
+ "create pipe p4"
+ + " with extractor ()"
+ + " with processor ()"
+ + " with connector ("
+ + "'connector'='invalid-param',"
+ + "'connector.ip'='%s',"
+ + "'connector.port'='%s',"
+ + "'connector.batch.enable'='false')",
+ receiverIp, receiverPort));
+ fail();
+ } catch (SQLException ignored) {
+ }
+
+ final List<TShowPipeInfo> showPipeResult = client.showPipe(new TShowPipeReq()).pipeInfoList;
+ Assert.assertEquals(1, showPipeResult.size());
+ }
+ }
+
+ @Test
+ public void testBrackets() throws Exception {
+ final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+
+ final String receiverIp = receiverDataNode.getIp();
+ final int receiverPort = receiverDataNode.getPort();
+
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+ try (final Connection connection = senderEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ final Statement statement = connection.createStatement()) {
+ statement.execute(
+ String.format(
+ "create pipe extractor1"
+ + " with extractor ('extractor'='iotdb-extractor')"
+ + " with connector ("
+ + "'connector'='iotdb-thrift-connector',"
+ + "'connector.ip'='%s',"
+ + "'connector.port'='%s',"
+ + "'connector.batch.enable'='false')",
+ receiverIp, receiverPort));
+ } catch (SQLException e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+
+ try (final Connection connection = senderEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ final Statement statement = connection.createStatement()) {
+ statement.execute(
+ String.format(
+ "create pipe source2"
+ + " with extractor (\"extractor\"=\"iotdb-extractor\")"
+ + " with connector ("
+ + "'connector'='iotdb-thrift-connector',"
+ + "'connector.ip'='%s',"
+ + "'connector.port'='%s',"
+ + "'connector.batch.enable'='false')",
+ receiverIp, receiverPort));
+ fail();
+ } catch (SQLException e) {
+ e.printStackTrace();
+ }
+
+ try (final Connection connection = senderEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ final Statement statement = connection.createStatement()) {
+ statement.execute(
+ String.format(
+ "create pipe extractor3"
+ + " with extractor ("
+ + "'capture.table'='true',"
+ + "'database-name'='test',"
+ + "'table-name'='test',"
+ + "'mode.streaming'='true',"
+ + "'mode.strict'='true',"
+ + "'mode.snapshot'='false',"
+ + "'start-time'='1',"
+ + "'end-time'='2',"
+ + "'extractor'=\"iotdb-extractor\")"
+ + " with connector ("
+ + "'connector'='iotdb-thrift-connector',"
+ + "'connector.ip'='%s',"
+ + "'connector.port'='%s',"
+ + "'connector.batch.enable'='false')",
+ receiverIp, receiverPort));
+ fail();
+ } catch (SQLException e) {
+ e.printStackTrace();
+ }
+
+ try (final Connection connection = senderEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ final Statement statement = connection.createStatement()) {
+ statement.execute(
+ String.format(
+ "create pipe extractor4"
+ + " with extractor ("
+ + "'extractor'=iotdb-extractor,"
+ + "'capture.table'='true',"
+ + "'database-name'='test',"
+ + "'table-name'='test',"
+ + "'mode.streaming'='true',"
+ + "'mode.strict'='true',"
+ + "'mode.snapshot'='false',"
+ + "'start-time'='1',"
+ + "'end-time'='2')"
+ + " with connector ("
+ + "'connector'='iotdb-thrift-connector',"
+ + "'connector.ip'='%s',"
+ + "'connector.port'='%s',"
+ + "'connector.batch.enable'='false')",
+ receiverIp, receiverPort));
+ fail();
+ } catch (SQLException ignored) {
+ }
+
+ try (final Connection connection = senderEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ final Statement statement = connection.createStatement()) {
+ statement.execute(
+ String.format(
+ "create pipe extractor5"
+ + " with extractor ("
+ + "'extractor'=`iotdb-extractor`,"
+ + "'capture.table'='true',"
+ + "'database-name'='test',"
+ + "'table-name'='test',"
+ + "'mode.streaming'='true',"
+ + "'mode.strict'='true',"
+ + "'mode.snapshot'='false',"
+ + "'start-time'='1',"
+ + "'end-time'='2')"
+ + " with connector ("
+ + "'connector'='iotdb-thrift-connector',"
+ + "'connector.ip'='%s',"
+ + "'connector.port'='%s',"
+ + "'connector.batch.enable'='false')",
+ receiverIp, receiverPort));
+ fail();
+ } catch (SQLException ignored) {
+ }
+
+ try (final Connection connection = senderEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ final Statement statement = connection.createStatement()) {
+ statement.execute(
+ String.format(
+ "create pipe processor1"
+ + " with processor ('processor'='do-nothing-processor')"
+ + " with connector ("
+ + "'connector'='iotdb-thrift-connector',"
+ + "'connector.ip'='%s',"
+ + "'connector.port'='%s',"
+ + "'connector.batch.enable'='false')",
+ receiverIp, receiverPort));
+ } catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+
+ try (final Connection connection = senderEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ final Statement statement = connection.createStatement()) {
+ statement.execute(
+ String.format(
+ "create pipe processor2"
+ + " with processor (\"processor\"=\"do-nothing-processor\")"
+ + " with connector ("
+ + "'connector'='iotdb-thrift-connector',"
+ + "'connector.ip'='%s',"
+ + "'connector.port'='%s',"
+ + "'connector.batch.enable'='false')",
+ receiverIp, receiverPort));
+ fail();
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+
+ try (final Connection connection = senderEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ final Statement statement = connection.createStatement()) {
+ statement.execute(
+ String.format(
+ "create pipe processor3"
+ + " with processor ('processor'=\"do-nothing-processor\")"
+ + " with connector ("
+ + "'connector'='iotdb-thrift-connector',"
+ + "'connector.ip'='%s',"
+ + "'connector.port'='%s',"
+ + "'connector.batch.enable'='false')",
+ receiverIp, receiverPort));
+ fail();
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+
+ try (final Connection connection = senderEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ final Statement statement = connection.createStatement()) {
+ statement.execute(
+ String.format(
+ "create pipe processor4"
+ + " with processor (processor=do-nothing-processor)"
+ + " with connector ("
+ + "'connector'='iotdb-thrift-connector',"
+ + "'connector.ip'='%s',"
+ + "'connector.port'='%s',"
+ + "'connector.batch.enable'='false')",
+ receiverIp, receiverPort));
+ fail();
+ } catch (Exception ignored) {
+ }
+
+ try (final Connection connection = senderEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ final Statement statement = connection.createStatement()) {
+ statement.execute(
+ String.format(
+ "create pipe processor5"
+ + " with processor ('processor'=`do-nothing-processor`)"
+ + " with connector ("
+ + "'connector'='iotdb-thrift-connector',"
+ + "'connector.ip'='%s',"
+ + "'connector.port'='%s',"
+ + "'connector.batch.enable'='false')",
+ receiverIp, receiverPort));
+ fail();
+ } catch (Exception ignored) {
+ }
+
+ final List<TShowPipeInfo> showPipeResult = client.showPipe(new TShowPipeReq()).pipeInfoList;
+ Assert.assertEquals(2, showPipeResult.size());
+ }
+ }
+
+ @Test
+ public void testShowPipeWithWrongPipeName() throws Exception {
+ final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+
+ final String receiverIp = receiverDataNode.getIp();
+ final int receiverPort = receiverDataNode.getPort();
+
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+ final Map<String, String> extractorAttributes = new HashMap<>();
+ final Map<String, String> processorAttributes = new HashMap<>();
+ final Map<String, String> connectorAttributes = new HashMap<>();
+
+ extractorAttributes.put("extractor.database-name", "test");
+ extractorAttributes.put("extractor.table-name", "test.*");
+ extractorAttributes.put("extractor.inclusion", "data.insert");
+ extractorAttributes.put("extractor.capture.table", "true");
+ extractorAttributes.put("extractor.mode.streaming", "true");
+ extractorAttributes.put("extractor.mode.snapshot", "true");
+ extractorAttributes.put("extractor.mode.strict", "true");
+ extractorAttributes.put("extractor.start-time", "1");
+ extractorAttributes.put("extractor.end-time", "2");
+
+ connectorAttributes.put("connector", "iotdb-thrift-connector");
+ connectorAttributes.put("connector.batch.enable", "false");
+ connectorAttributes.put("connector.ip", receiverIp);
+ connectorAttributes.put("connector.port", Integer.toString(receiverPort));
+
+ TSStatus status =
+ client.createPipe(
+ new TCreatePipeReq("p1", connectorAttributes)
+ .setExtractorAttributes(extractorAttributes)
+ .setProcessorAttributes(processorAttributes));
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+
+ status =
+ client.createPipe(
+ new TCreatePipeReq("p2", connectorAttributes)
+ .setExtractorAttributes(extractorAttributes)
+ .setProcessorAttributes(processorAttributes));
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+
+ connectorAttributes.replace("connector.batch.enable", "true");
+
+ status =
+ client.createPipe(
+ new TCreatePipeReq("p3", connectorAttributes)
+ .setExtractorAttributes(extractorAttributes)
+ .setProcessorAttributes(processorAttributes));
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+
+ List<TShowPipeInfo> showPipeResult = client.showPipe(new TShowPipeReq()).pipeInfoList;
+ Assert.assertEquals(3, showPipeResult.size());
+
+ showPipeResult = client.showPipe(new TShowPipeReq().setPipeName("p1")).pipeInfoList;
+ Assert.assertTrue(showPipeResult.stream().anyMatch((o) -> o.id.equals("p1")));
+ Assert.assertFalse(showPipeResult.stream().anyMatch((o) -> o.id.equals("p2")));
+ Assert.assertFalse(showPipeResult.stream().anyMatch((o) -> o.id.equals("p3")));
+
+ // Show all pipes whose connector is also used by p1.
+ // p1 and p2 share the same connector parameters, so they have the same connector.
+ showPipeResult =
+ client.showPipe(new TShowPipeReq().setPipeName("p1").setWhereClause(true)).pipeInfoList;
+ Assert.assertTrue(showPipeResult.stream().anyMatch((o) -> o.id.equals("p1")));
+ Assert.assertTrue(showPipeResult.stream().anyMatch((o) -> o.id.equals("p2")));
+ Assert.assertFalse(showPipeResult.stream().anyMatch((o) -> o.id.equals("p3")));
+ }
+ }
+
+ @Test
+ public void testInclusionPattern() throws Exception {
+ final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+
+ final String receiverIp = receiverDataNode.getIp();
+ final int receiverPort = receiverDataNode.getPort();
+
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+ // Empty inclusion
+ try (final Connection connection = senderEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ final Statement statement = connection.createStatement()) {
+ statement.execute(
+ String.format(
+ "create pipe p2"
+ + " with extractor ('extractor.inclusion'='schema, auth.role', 'extractor.inclusion.exclusion'='all')"
+ + " with processor ()"
+ + " with connector ("
+ + "'connector'='iotdb-thrift-connector',"
+ + "'connector.ip'='%s',"
+ + "'connector.port'='%s',"
+ + "'connector.batch.enable'='false')",
+ receiverIp, receiverPort));
+ fail();
+ } catch (SQLException ignored) {
+ }
+
+ // Invalid inclusion
+ try (final Connection connection = senderEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ final Statement statement = connection.createStatement()) {
+ statement.execute(
+ String.format(
+ "create pipe p3"
+ + " with extractor ('extractor.inclusion'='wrong')"
+ + " with processor ()"
+ + " with connector ("
+ + "'connector'='iotdb-thrift-connector',"
+ + "'connector.ip'='%s',"
+ + "'connector.port'='%s',"
+ + "'connector.batch.enable'='false')",
+ receiverIp, receiverPort));
+ fail();
+ } catch (SQLException ignored) {
+ }
+
+ // Invalid exclusion
+ try (final Connection connection = senderEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ final Statement statement = connection.createStatement()) {
+ statement.execute(
+ String.format(
+ "create pipe p4"
+ + " with extractor ('extractor.inclusion.exclusion'='wrong')"
+ + " with processor ()"
+ + " with connector ("
+ + "'connector'='iotdb-thrift-connector',"
+ + "'connector.ip'='%s',"
+ + "'connector.port'='%s',"
+ + "'connector.batch.enable'='false')",
+ receiverIp, receiverPort));
+ fail();
+ } catch (SQLException ignored) {
+ }
+
+ // Valid
+ try (final Connection connection = senderEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ final Statement statement = connection.createStatement()) {
+ statement.execute(
+ String.format(
+ "create pipe p4"
+ + " with extractor ('extractor.inclusion'='all', 'extractor.inclusion.exclusion'='schema, auth')"
+ + " with processor ()"
+ + " with connector ("
+ + "'connector'='iotdb-thrift-connector',"
+ + "'connector.ip'='%s',"
+ + "'connector.port'='%s',"
+ + "'connector.batch.enable'='false')",
+ receiverIp, receiverPort));
+ } catch (SQLException e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+
+ final List<TShowPipeInfo> showPipeResult = client.showPipe(new TShowPipeReq()).pipeInfoList;
+ Assert.assertEquals(1, showPipeResult.size());
+ }
+ }
+}
diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/tablemodel/IoTDBPipeWithLoadIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/tablemodel/IoTDBPipeWithLoadIT.java
new file mode 100644
index 00000000000..77dffd9f6db
--- /dev/null
+++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/tablemodel/IoTDBPipeWithLoadIT.java
@@ -0,0 +1,120 @@
+/*
+ * 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.pipe.it.tablemodel;
+
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.client.sync.SyncConfigNodeIServiceClient;
+import org.apache.iotdb.confignode.rpc.thrift.TCreatePipeReq;
+import org.apache.iotdb.consensus.ConsensusFactory;
+import org.apache.iotdb.it.env.MultiEnvFactory;
+import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper;
+import org.apache.iotdb.it.framework.IoTDBTestRunner;
+import org.apache.iotdb.itbase.category.MultiClusterIT2TableModel;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Ignore;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import java.util.HashMap;
+import java.util.Map;
+
+@RunWith(IoTDBTestRunner.class)
+@Category({MultiClusterIT2TableModel.class})
+public class IoTDBPipeWithLoadIT extends AbstractPipeTableModelTestIT {
+
+ @Before
+ public void setUp() {
+ MultiEnvFactory.createEnv(2);
+ senderEnv = MultiEnvFactory.getEnv(0);
+ receiverEnv = MultiEnvFactory.getEnv(1);
+
+ // TODO: delete ratis configurations
+ senderEnv
+ .getConfig()
+ .getCommonConfig()
+ .setAutoCreateSchemaEnabled(true)
+ .setConfigNodeConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS)
+ .setSchemaRegionConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS)
+ // Disable sender compaction to test mods
+ .setEnableSeqSpaceCompaction(false)
+ .setEnableUnseqSpaceCompaction(false)
+ .setEnableCrossSpaceCompaction(false);
+ receiverEnv
+ .getConfig()
+ .getCommonConfig()
+ .setAutoCreateSchemaEnabled(true)
+ .setConfigNodeConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS)
+ .setSchemaRegionConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS);
+
+ // 10 min, assert that the operations will not time out
+ senderEnv.getConfig().getCommonConfig().setCnConnectionTimeoutMs(600000);
+ receiverEnv.getConfig().getCommonConfig().setCnConnectionTimeoutMs(600000);
+
+ senderEnv.initClusterEnvironment();
+ receiverEnv.initClusterEnvironment();
+ }
+
+ @Ignore // not support
+ @Test
+ public void testReceiverNotLoadDeletedTimeseries() throws Exception {
+ final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+ final String receiverIp = receiverDataNode.getIp();
+ final int receiverPort = receiverDataNode.getPort();
+ boolean insertResult = true;
+
+ final Map<String, String> extractorAttributes = new HashMap<>();
+ final Map<String, String> processorAttributes = new HashMap<>();
+ final Map<String, String> connectorAttributes = new HashMap<>();
+
+ // Enable mods transfer
+ extractorAttributes.put("modes", "true");
+ extractorAttributes.put("capture.table", "true");
+
+ connectorAttributes.put("connector.batch.enable", "false");
+ connectorAttributes.put("connector.ip", receiverIp);
+ connectorAttributes.put("connector.port", Integer.toString(receiverPort));
+
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+ // Generate TsFile
+ TableModelUtils.createDataBaseAndTable(senderEnv, "test", "test");
+ insertResult = TableModelUtils.insertData("test", "test", 0, 100, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ TableModelUtils.deleteData("test", "test", 50, 100, senderEnv);
+
+ TSStatus status =
+ client.createPipe(
+ new TCreatePipeReq("p1", connectorAttributes)
+ .setExtractorAttributes(extractorAttributes)
+ .setProcessorAttributes(processorAttributes));
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("p1").getCode());
+
+ TableModelUtils.assertCountData("test", "test", 50, receiverEnv);
+ }
+ }
+}
diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/tablemodel/IoTDBTablePatternFormatIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/tablemodel/IoTDBTablePatternFormatIT.java
new file mode 100644
index 00000000000..62b66aba5fe
--- /dev/null
+++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/tablemodel/IoTDBTablePatternFormatIT.java
@@ -0,0 +1,572 @@
+/*
+ * 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.pipe.it.tablemodel;
+
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.client.sync.SyncConfigNodeIServiceClient;
+import org.apache.iotdb.confignode.rpc.thrift.TCreatePipeReq;
+import org.apache.iotdb.db.it.utils.TestUtils;
+import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper;
+import org.apache.iotdb.it.framework.IoTDBTestRunner;
+import org.apache.iotdb.itbase.category.MultiClusterIT2TableModel;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+
+@RunWith(IoTDBTestRunner.class)
+@Category({MultiClusterIT2TableModel.class})
+public class IoTDBTablePatternFormatIT extends AbstractPipeTableModelTestIT {
+
+ @Test
+ public void testTableNamePattern() throws Exception {
+ final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+
+ final String receiverIp = receiverDataNode.getIp();
+ final int receiverPort = receiverDataNode.getPort();
+ boolean insertResult = true;
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+
+ TableModelUtils.createDataBaseAndTable(senderEnv, "test", "test");
+ TableModelUtils.createDataBaseAndTable(senderEnv, "test1", "test1");
+ TableModelUtils.createDataBaseAndTable(senderEnv, "pattern", "pattern");
+ TableModelUtils.createDataBaseAndTable(senderEnv, "pattern1", "pattern1");
+ insertResult = TableModelUtils.insertData("test", "test", 0, 100, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ insertResult = TableModelUtils.insertData("test1", "test1", 0, 100, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ insertResult = TableModelUtils.insertData("pattern", "pattern", 0, 100, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ insertResult = TableModelUtils.insertData("pattern1", "pattern1", 0, 100, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+
+ final Map<String, String> extractorAttributes = new HashMap<>();
+ final Map<String, String> processorAttributes = new HashMap<>();
+ final Map<String, String> connectorAttributes = new HashMap<>();
+
+ extractorAttributes.put("extractor.table-name", "test.*");
+ extractorAttributes.put("extractor.inclusion", "data.insert");
+ extractorAttributes.put("extractor.capture.table", "true");
+
+ connectorAttributes.put("connector", "iotdb-thrift-connector");
+ connectorAttributes.put("connector.batch.enable", "false");
+ connectorAttributes.put("connector.ip", receiverIp);
+ connectorAttributes.put("connector.port", Integer.toString(receiverPort));
+
+ final TSStatus status =
+ client.createPipe(
+ new TCreatePipeReq("p1", connectorAttributes)
+ .setExtractorAttributes(extractorAttributes)
+ .setProcessorAttributes(processorAttributes));
+
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("p1").getCode());
+ insertResult = TableModelUtils.insertData("test", "test", 100, 200, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ insertResult = TableModelUtils.insertData("test1", "test1", 100, 200, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ insertResult = TableModelUtils.insertData("pattern", "pattern", 100, 200, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ insertResult = TableModelUtils.insertData("pattern1", "pattern1", 100, 200, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+
+ TableModelUtils.assertData("test", "test", 0, 200, receiverEnv);
+ TableModelUtils.assertData("test1", "test1", 0, 200, receiverEnv);
+ if (!TableModelUtils.hasDataBase("test", receiverEnv)) {
+ Assert.fail();
+ }
+ if (!TableModelUtils.hasDataBase("test1", receiverEnv)) {
+ Assert.fail();
+ }
+ }
+ }
+
+ @Test
+ public void testTableNamePatternByHistoryTSFile() throws Exception {
+ final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+
+ final String receiverIp = receiverDataNode.getIp();
+ final int receiverPort = receiverDataNode.getPort();
+ boolean insertResult = true;
+
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+
+ TableModelUtils.createDataBaseAndTable(senderEnv, "test", "test");
+ TableModelUtils.createDataBaseAndTable(senderEnv, "test1", "test1");
+ TableModelUtils.createDataBaseAndTable(senderEnv, "pattern", "pattern");
+ TableModelUtils.createDataBaseAndTable(senderEnv, "pattern1", "pattern1");
+ insertResult = TableModelUtils.insertData("test", "test", 0, 100, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ insertResult = TableModelUtils.insertData("test1", "test1", 0, 100, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ insertResult = TableModelUtils.insertData("pattern", "pattern", 0, 100, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ insertResult = TableModelUtils.insertData("pattern1", "pattern1", 0, 100, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+
+ final Map<String, String> extractorAttributes = new HashMap<>();
+ final Map<String, String> processorAttributes = new HashMap<>();
+ final Map<String, String> connectorAttributes = new HashMap<>();
+
+ // extractorAttributes.put("extractor.database-name", "test.*");
+ extractorAttributes.put("extractor.table-name", "test.*");
+ extractorAttributes.put("extractor.inclusion", "data.insert");
+ extractorAttributes.put("extractor.capture.table", "true");
+ extractorAttributes.put("extractor.start-time", "0");
+ extractorAttributes.put("extractor.end-time", "49");
+
+ connectorAttributes.put("connector", "iotdb-thrift-connector");
+ connectorAttributes.put("connector.batch.enable", "false");
+ connectorAttributes.put("connector.ip", receiverIp);
+ connectorAttributes.put("connector.port", Integer.toString(receiverPort));
+
+ final TSStatus status =
+ client.createPipe(
+ new TCreatePipeReq("p1", connectorAttributes)
+ .setExtractorAttributes(extractorAttributes)
+ .setProcessorAttributes(processorAttributes));
+
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("p1").getCode());
+
+ TableModelUtils.assertData("test", "test", 0, 50, receiverEnv);
+ TableModelUtils.assertData("test1", "test1", 0, 50, receiverEnv);
+ HashSet<String> expectedResults = new HashSet();
+ if (!TableModelUtils.hasDataBase("test", receiverEnv)) {
+ Assert.fail();
+ }
+ if (!TableModelUtils.hasDataBase("test1", receiverEnv)) {
+ Assert.fail();
+ }
+ }
+ }
+
+ @Test
+ public void testTableNamePatternByRealTime() throws Exception {
+ final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+
+ final String receiverIp = receiverDataNode.getIp();
+ final int receiverPort = receiverDataNode.getPort();
+ boolean insertResult = true;
+
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+
+ TableModelUtils.createDataBaseAndTable(senderEnv, "test", "test");
+ TableModelUtils.createDataBaseAndTable(senderEnv, "test1", "test1");
+ TableModelUtils.createDataBaseAndTable(senderEnv, "pattern", "pattern");
+ TableModelUtils.createDataBaseAndTable(senderEnv, "pattern1", "pattern1");
+
+ final Map<String, String> extractorAttributes = new HashMap<>();
+ final Map<String, String> processorAttributes = new HashMap<>();
+ final Map<String, String> connectorAttributes = new HashMap<>();
+
+ // extractorAttributes.put("extractor.database-name", "test.*");
+ extractorAttributes.put("extractor.table-name", "test.*");
+ extractorAttributes.put("extractor.inclusion", "data.insert");
+ extractorAttributes.put("extractor.capture.table", "true");
+ extractorAttributes.put("extractor.start-time", "100");
+ extractorAttributes.put("extractor.end-time", "149");
+
+ connectorAttributes.put("connector", "iotdb-thrift-connector");
+ connectorAttributes.put("connector.batch.enable", "false");
+ connectorAttributes.put("connector.ip", receiverIp);
+ connectorAttributes.put("connector.port", Integer.toString(receiverPort));
+
+ final TSStatus status =
+ client.createPipe(
+ new TCreatePipeReq("p1", connectorAttributes)
+ .setExtractorAttributes(extractorAttributes)
+ .setProcessorAttributes(processorAttributes));
+
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("p1").getCode());
+ insertResult = TableModelUtils.insertData("test", "test", 100, 200, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ insertResult = TableModelUtils.insertData("test1", "test1", 100, 200, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ insertResult = TableModelUtils.insertData("pattern", "pattern", 100, 200, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ insertResult = TableModelUtils.insertData("pattern1", "pattern1", 100, 200, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+
+ TableModelUtils.assertData("test", "test", 100, 150, receiverEnv);
+ TableModelUtils.assertData("test1", "test1", 100, 150, receiverEnv);
+ if (!TableModelUtils.hasDataBase("test", receiverEnv)) {
+ Assert.fail();
+ }
+ if (!TableModelUtils.hasDataBase("test1", receiverEnv)) {
+ Assert.fail();
+ }
+ }
+ }
+
+ @Test
+ public void testDataBasePattern() throws Exception {
+ final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+
+ final String receiverIp = receiverDataNode.getIp();
+ final int receiverPort = receiverDataNode.getPort();
+ boolean insertResult = true;
+
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+
+ TableModelUtils.createDataBaseAndTable(senderEnv, "test", "test");
+ TableModelUtils.createDataBaseAndTable(senderEnv, "test1", "test1");
+ TableModelUtils.createDataBaseAndTable(senderEnv, "pattern", "pattern");
+ TableModelUtils.createDataBaseAndTable(senderEnv, "pattern1", "pattern1");
+
+ insertResult = TableModelUtils.insertData("test", "test", 0, 100, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ insertResult = TableModelUtils.insertData("test1", "test1", 0, 100, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ insertResult = TableModelUtils.insertData("pattern", "pattern", 0, 100, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ insertResult = TableModelUtils.insertData("pattern1", "pattern1", 0, 100, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+
+ final Map<String, String> extractorAttributes = new HashMap<>();
+ final Map<String, String> processorAttributes = new HashMap<>();
+ final Map<String, String> connectorAttributes = new HashMap<>();
+
+ extractorAttributes.put("extractor.database-name", "pattern.*");
+ extractorAttributes.put("extractor.inclusion", "data.insert");
+ extractorAttributes.put("extractor.capture.table", "true");
+
+ connectorAttributes.put("connector", "iotdb-thrift-connector");
+ connectorAttributes.put("connector.batch.enable", "false");
+ connectorAttributes.put("connector.ip", receiverIp);
+ connectorAttributes.put("connector.port", Integer.toString(receiverPort));
+
+ final TSStatus status =
+ client.createPipe(
+ new TCreatePipeReq("p1", connectorAttributes)
+ .setExtractorAttributes(extractorAttributes)
+ .setProcessorAttributes(processorAttributes));
+
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("p1").getCode());
+ insertResult = TableModelUtils.insertData("test", "test", 100, 200, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ insertResult = TableModelUtils.insertData("test1", "test1", 100, 200, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ insertResult = TableModelUtils.insertData("pattern", "pattern", 100, 200, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ insertResult = TableModelUtils.insertData("pattern1", "pattern1", 100, 200, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ TableModelUtils.assertData("pattern", "pattern", 0, 200, receiverEnv);
+ TableModelUtils.assertData("pattern1", "pattern1", 0, 200, receiverEnv);
+
+ HashSet<String> expectedResults = new HashSet();
+ if (!TableModelUtils.hasDataBase("pattern", receiverEnv)) {
+ Assert.fail();
+ }
+ if (!TableModelUtils.hasDataBase("pattern1", receiverEnv)) {
+ Assert.fail();
+ }
+ }
+ }
+
+ @Test
+ public void testDataBasePatternByHistoryTSFile() throws Exception {
+ final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+
+ final String receiverIp = receiverDataNode.getIp();
+ final int receiverPort = receiverDataNode.getPort();
+ boolean insertResult = true;
+
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+
+ TableModelUtils.createDataBaseAndTable(senderEnv, "test", "test");
+ TableModelUtils.createDataBaseAndTable(senderEnv, "test1", "test1");
+ TableModelUtils.createDataBaseAndTable(senderEnv, "pattern", "pattern");
+ TableModelUtils.createDataBaseAndTable(senderEnv, "pattern1", "pattern1");
+
+ insertResult = TableModelUtils.insertData("test", "test", 0, 100, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ insertResult = TableModelUtils.insertData("test1", "test1", 0, 100, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ insertResult = TableModelUtils.insertData("pattern", "pattern", 0, 100, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ insertResult = TableModelUtils.insertData("pattern1", "pattern1", 0, 100, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+
+ final Map<String, String> extractorAttributes = new HashMap<>();
+ final Map<String, String> processorAttributes = new HashMap<>();
+ final Map<String, String> connectorAttributes = new HashMap<>();
+
+ extractorAttributes.put("extractor.database-name", "pattern.*");
+ extractorAttributes.put("extractor.inclusion", "data.insert");
+ extractorAttributes.put("extractor.capture.table", "true");
+
+ connectorAttributes.put("connector", "iotdb-thrift-connector");
+ connectorAttributes.put("connector.batch.enable", "false");
+ connectorAttributes.put("connector.ip", receiverIp);
+ connectorAttributes.put("connector.port", Integer.toString(receiverPort));
+
+ final TSStatus status =
+ client.createPipe(
+ new TCreatePipeReq("p1", connectorAttributes)
+ .setExtractorAttributes(extractorAttributes)
+ .setProcessorAttributes(processorAttributes));
+
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("p1").getCode());
+ TableModelUtils.assertData("pattern", "pattern", 0, 100, receiverEnv);
+ TableModelUtils.assertData("pattern1", "pattern1", 0, 100, receiverEnv);
+
+ if (!TableModelUtils.hasDataBase("pattern1", receiverEnv)) {
+ Assert.fail();
+ }
+ if (!TableModelUtils.hasDataBase("pattern", receiverEnv)) {
+ Assert.fail();
+ }
+ }
+ }
+
+ @Test
+ public void testDataBasePatternByRealtime() throws Exception {
+ final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+
+ final String receiverIp = receiverDataNode.getIp();
+ final int receiverPort = receiverDataNode.getPort();
+ boolean insertResult = true;
+
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+
+ TableModelUtils.createDataBaseAndTable(senderEnv, "test", "test");
+ TableModelUtils.createDataBaseAndTable(senderEnv, "test1", "test1");
+ TableModelUtils.createDataBaseAndTable(senderEnv, "pattern", "pattern");
+ TableModelUtils.createDataBaseAndTable(senderEnv, "pattern1", "pattern1");
+
+ final Map<String, String> extractorAttributes = new HashMap<>();
+ final Map<String, String> processorAttributes = new HashMap<>();
+ final Map<String, String> connectorAttributes = new HashMap<>();
+
+ extractorAttributes.put("extractor.database-name", "pattern.*");
+ extractorAttributes.put("extractor.inclusion", "data.insert");
+ extractorAttributes.put("extractor.capture.table", "true");
+
+ connectorAttributes.put("connector", "iotdb-thrift-connector");
+ connectorAttributes.put("connector.batch.enable", "false");
+ connectorAttributes.put("connector.ip", receiverIp);
+ connectorAttributes.put("connector.port", Integer.toString(receiverPort));
+
+ final TSStatus status =
+ client.createPipe(
+ new TCreatePipeReq("p1", connectorAttributes)
+ .setExtractorAttributes(extractorAttributes)
+ .setProcessorAttributes(processorAttributes));
+
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("p1").getCode());
+
+ insertResult = TableModelUtils.insertData("pattern", "pattern", 0, 100, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ insertResult = TableModelUtils.insertData("pattern1", "pattern1", 0, 100, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ insertResult = TableModelUtils.insertData("test", "test", 0, 100, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ insertResult = TableModelUtils.insertData("test1", "test1", 0, 100, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+
+ TableModelUtils.assertData("pattern", "pattern", 0, 100, receiverEnv);
+ TableModelUtils.assertData("pattern1", "pattern1", 0, 100, receiverEnv);
+
+ if (!TableModelUtils.hasDataBase("pattern", receiverEnv)) {
+ Assert.fail();
+ }
+ if (!TableModelUtils.hasDataBase("pattern1", receiverEnv)) {
+ Assert.fail();
+ }
+ }
+ }
+
+ @Test
+ public void testIoTDBPatternWithDataBaseAndTable() throws Exception {
+ final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+
+ final String receiverIp = receiverDataNode.getIp();
+ final int receiverPort = receiverDataNode.getPort();
+ boolean insertResult = true;
+
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+
+ TableModelUtils.createDataBaseAndTable(senderEnv, "test", "test");
+ TableModelUtils.createDataBaseAndTable(senderEnv, "test1", "test1");
+ TableModelUtils.createDataBaseAndTable(senderEnv, "pattern", "pattern");
+ TableModelUtils.createDataBaseAndTable(senderEnv, "pattern1", "pattern1");
+
+ insertResult = TableModelUtils.insertData("test", "test", 0, 100, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ insertResult = TableModelUtils.insertData("test1", "test1", 0, 100, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ insertResult = TableModelUtils.insertData("pattern", "pattern", 0, 100, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ insertResult = TableModelUtils.insertData("pattern1", "pattern1", 0, 100, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+
+ final Map<String, String> extractorAttributes = new HashMap<>();
+ final Map<String, String> processorAttributes = new HashMap<>();
+ final Map<String, String> connectorAttributes = new HashMap<>();
+
+ extractorAttributes.put("extractor.database-name", "pattern.*");
+ extractorAttributes.put("extractor.table-name", "test.*");
+ extractorAttributes.put("extractor.inclusion", "data.insert");
+ extractorAttributes.put("extractor.capture.table", "true");
+
+ connectorAttributes.put("connector", "iotdb-thrift-connector");
+ connectorAttributes.put("connector.batch.enable", "false");
+ connectorAttributes.put("connector.ip", receiverIp);
+ connectorAttributes.put("connector.port", Integer.toString(receiverPort));
+
+ final TSStatus status =
+ client.createPipe(
+ new TCreatePipeReq("p1", connectorAttributes)
+ .setExtractorAttributes(extractorAttributes)
+ .setProcessorAttributes(processorAttributes));
+
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("p1").getCode());
+ insertResult = TableModelUtils.insertData("test", "test", 100, 200, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ insertResult = TableModelUtils.insertData("test1", "test1", 100, 200, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ insertResult = TableModelUtils.insertData("pattern", "pattern", 100, 200, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+ insertResult = TableModelUtils.insertData("pattern1", "pattern1", 100, 200, senderEnv);
+ if (!insertResult) {
+ return;
+ }
+
+ TestUtils.assertDataEventuallyOnEnv(
+ receiverEnv,
+ "show databases",
+ "Database,SchemaReplicationFactor,DataReplicationFactor,TimePartitionInterval,",
+ Collections.emptySet(),
+ null);
+ }
+ }
+}
diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/tablemodel/TableModelUtils.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/tablemodel/TableModelUtils.java
new file mode 100644
index 00000000000..4c8ff89b024
--- /dev/null
+++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/tablemodel/TableModelUtils.java
@@ -0,0 +1,395 @@
+/*
+ * 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.pipe.it.tablemodel;
+
+import org.apache.iotdb.db.it.utils.TestUtils;
+import org.apache.iotdb.isession.IPooledSession;
+import org.apache.iotdb.isession.pool.ISessionPool;
+import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper;
+import org.apache.iotdb.itbase.env.BaseEnv;
+import org.apache.iotdb.rpc.RpcUtils;
+
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.utils.Binary;
+import org.apache.tsfile.utils.BitMap;
+import org.apache.tsfile.utils.BytesUtils;
+import org.apache.tsfile.utils.DateUtils;
+import org.apache.tsfile.write.record.Tablet;
+import org.apache.tsfile.write.schema.IMeasurementSchema;
+import org.apache.tsfile.write.schema.MeasurementSchema;
+
+import java.nio.charset.StandardCharsets;
+import java.sql.Connection;
+import java.sql.Statement;
+import java.text.SimpleDateFormat;
+import java.time.LocalDate;
+import java.time.ZoneOffset;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Date;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Random;
+import java.util.Set;
+
+import static org.junit.Assert.fail;
+
+/**
+ * This class is specifically designed for table model testing and is used for data insertion.
+ * Please note that the Table pattern defined by this class is fixed, and it is recommended to use
+ * all methods together to ensure data consistency and integrity.
+ *
+ * <p>This class provides a structured approach to inserting data, suitable for scenarios where
+ * batch data insertion is required. Due to the fixed pattern, users should ensure that the data
+ * format matches the pattern defined by the class.
+ */
+public class TableModelUtils {
+
+ public static void createDataBaseAndTable(BaseEnv baseEnv, String table, String database) {
+ try (Connection connection = baseEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ Statement statement = connection.createStatement()) {
+ statement.execute("create database if not exists " + database);
+ statement.execute("use " + database);
+ statement.execute(
+ "CREATE TABLE "
+ + table
+ + "(s0 string id, s1 int64 measurement, s2 float measurement, s3 string measurement, s4 timestamp measurement, s5 int32 measurement, s6 double measurement, s7 date measurement, s8 text measurement )");
+ } catch (Exception e) {
+ fail(e.getMessage());
+ }
+ }
+
+ public static void createDataBase(BaseEnv baseEnv, String database) {
+ try (Connection connection = baseEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ Statement statement = connection.createStatement()) {
+ statement.execute("create database if not exists " + database);
+ } catch (Exception e) {
+ fail(e.getMessage());
+ }
+ }
+
+ public static boolean insertData(
+ String dataBaseName, String tableName, int start, int end, BaseEnv baseEnv) {
+ List<String> list = new ArrayList<>(end - start + 1);
+ for (int i = start; i < end; ++i) {
+ list.add(
+ String.format(
+ "insert into %s (s0, s3, s2, s1, s4, s5, s6, s7, s8, time) values ('t%s','%s', %s.0, %s, %s, %d, %d.0, '%s', '%s', %s)",
+ tableName, i, i, i, i, i, i, i, getDateStr(i), i, i));
+ }
+ list.add("flush");
+ if (!TestUtils.tryExecuteNonQueriesWithRetry(
+ dataBaseName, BaseEnv.TABLE_SQL_DIALECT, baseEnv, list)) {
+ return false;
+ }
+ return true;
+ }
+
+ public static boolean insertData(
+ String dataBaseName,
+ String tableName,
+ int start,
+ int end,
+ BaseEnv baseEnv,
+ boolean allowNullValue) {
+ List<String> list = new ArrayList<>(end - start + 1);
+ Object[] values = new Object[9];
+ Random random = new Random();
+ // s0 string, s1 int64, s2 float, s3 string, s4 timestamp, s5 int32, s6 double, s7 date, s8 text
+ for (int i = start; i < end; ++i) {
+ Arrays.fill(values, i);
+ values[0] = String.format("'t%s'", i);
+ values[2] = String.format("%s.0", i);
+ values[3] = String.format("%s", i);
+ values[6] = String.format("%s.0", i);
+ values[7] = String.format("'%s'", getDateStr(i));
+ values[8] = String.format("'%s'", i);
+ if (allowNullValue) {
+ values[random.nextInt(9)] = "null";
+ }
+ list.add(
+ String.format(
+ "insert into %s (s0, s1, s2, s3, s4, s5, s6, s7, s8, time) values (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)",
+ tableName, values[0], values[1], values[2], values[3], values[4], values[5],
+ values[6], values[7], values[8], i));
+ }
+ list.add("flush");
+ if (!TestUtils.tryExecuteNonQueriesWithRetry(
+ dataBaseName, BaseEnv.TABLE_SQL_DIALECT, baseEnv, list)) {
+ return false;
+ }
+ return true;
+ }
+
+ public static boolean insertDataNotThrowError(
+ String dataBaseName, String tableName, int start, int end, BaseEnv baseEnv) {
+ List<String> list = new ArrayList<>(end - start + 1);
+ for (int i = start; i < end; ++i) {
+ list.add(
+ String.format(
+ "insert into %s (s0, s3, s2, s1, s4, s5, s6, s7, s8, time) values ('t%s','%s', %s.0, %s, %s, %d, %d.0, '%s', '%s', %s)",
+ tableName, i, i, i, i, i, i, i, getDateStr(i), i, i));
+ }
+ return TestUtils.tryExecuteNonQueriesWithRetry(
+ dataBaseName, BaseEnv.TABLE_SQL_DIALECT, baseEnv, list);
+ }
+
+ public static boolean insertData(
+ String dataBaseName,
+ String tableName,
+ int start,
+ int end,
+ BaseEnv baseEnv,
+ DataNodeWrapper wrapper) {
+ List<String> list = new ArrayList<>(end - start + 1);
+ for (int i = start; i < end; ++i) {
+ list.add(
+ String.format(
+ "insert into %s (s0, s3, s2, s1, s4, s5, s6, s7, s8, time) values ('t%s','%s', %s.0, %s, %s, %d, %d.0, '%s', '%s', %s)",
+ tableName, i, i, i, i, i, i, i, getDateStr(i), i, i));
+ }
+ list.add("flush");
+ if (!TestUtils.tryExecuteNonQueriesOnSpecifiedDataNodeWithRetry(
+ baseEnv, wrapper, list, dataBaseName, BaseEnv.TABLE_SQL_DIALECT)) {
+ return false;
+ }
+ return true;
+ }
+
+ public static boolean insertDataByTablet(
+ String dataBaseName,
+ String tableName,
+ int start,
+ int end,
+ BaseEnv baseEnv,
+ boolean allowNullValue) {
+ final Tablet tablet = generateTablet(tableName, start, end, allowNullValue);
+ ISessionPool sessionPool = baseEnv.getSessionPool(1, "table");
+ try (final IPooledSession session = sessionPool.getPooledSession()) {
+ session.executeNonQueryStatement("use " + dataBaseName);
+ session.insertTablet(tablet);
+ session.executeNonQueryStatement("flush");
+ return true;
+ } catch (Exception e) {
+ e.printStackTrace();
+ return false;
+ }
+ }
+
+ public static void deleteData(
+ String dataBaseName, String tableName, int start, int end, BaseEnv baseEnv) {
+ List<String> list = new ArrayList<>(end - start + 1);
+ list.add(
+ String.format("delete from %s where time >= %s and time <= %s", tableName, start, end));
+ list.add("flush");
+ if (!TestUtils.tryExecuteNonQueriesWithRetry(
+ dataBaseName, BaseEnv.TABLE_SQL_DIALECT, baseEnv, list)) {
+ fail();
+ }
+ }
+
+ public static Set<String> generateExpectedResults(int start, int end) {
+ Set<String> expectedResSet = new HashSet<>();
+ for (int i = start; i < end; ++i) {
+ final String time = RpcUtils.formatDatetime("default", "ms", i, ZoneOffset.UTC);
+ expectedResSet.add(
+ String.format(
+ "t%d,%d,%d.0,%d,%s,%d,%d.0,%s,%s,%s,",
+ i, i, i, i, time, i, i, getDateStr(i), i, time));
+ }
+ return expectedResSet;
+ }
+
+ public static Set<String> generateExpectedResults(Tablet tablet) {
+ Set<String> expectedResSet = new HashSet<>();
+ List<IMeasurementSchema> schemas = tablet.getSchemas();
+ for (int i = 0; i < tablet.rowSize; i++) {
+ StringBuilder stringBuffer = new StringBuilder();
+ for (int j = 0; j < tablet.getSchemas().size(); j++) {
+ BitMap bitMap = tablet.bitMaps[j];
+ if (bitMap.isMarked(i)) {
+ stringBuffer.append("null,");
+ continue;
+ }
+ switch (schemas.get(j).getType()) {
+ case TIMESTAMP:
+ final String time =
+ RpcUtils.formatDatetime(
+ "default", "ms", ((long[]) tablet.values[j])[i], ZoneOffset.UTC);
+ stringBuffer.append(time);
+ stringBuffer.append(",");
+ break;
+ case DATE:
+ stringBuffer.append(((LocalDate[]) tablet.values[j])[i].toString());
+ stringBuffer.append(",");
+ break;
+ case BLOB:
+ stringBuffer.append(
+ BytesUtils.parseBlobByteArrayToString(
+ ((Binary[]) tablet.values[j])[i].getValues()));
+ stringBuffer.append(",");
+ break;
+ case TEXT:
+ case STRING:
+ stringBuffer.append(
+ new String(((Binary[]) tablet.values[j])[i].getValues(), StandardCharsets.UTF_8));
+ stringBuffer.append(",");
+ break;
+ case DOUBLE:
+ stringBuffer.append(((double[]) tablet.values[j])[i]);
+ stringBuffer.append(",");
+ break;
+ case FLOAT:
+ stringBuffer.append(((float[]) tablet.values[j])[i]);
+ stringBuffer.append(",");
+ case INT32:
+ stringBuffer.append(((int[]) tablet.values[j])[i]);
+ stringBuffer.append(",");
+ case INT64:
+ stringBuffer.append(((long[]) tablet.values[j])[i]);
+ stringBuffer.append(",");
+ }
+ }
+ String time = RpcUtils.formatDatetime("default", "ms", tablet.timestamps[i], ZoneOffset.UTC);
+ stringBuffer.append(time);
+ stringBuffer.append(",");
+ expectedResSet.add(stringBuffer.toString());
+ }
+
+ return expectedResSet;
+ }
+
+ public static String generateHeaderResults() {
+ return "s0,s3,s2,s1,s4,s5,s6,s7,s8,time,";
+ }
+
+ public static String getQuerySql(String table) {
+ return "select s0,s3,s2,s1,s4,s5,s6,s7,s8,time from " + table;
+ }
+
+ public static String getQueryCountSql(String table) {
+ return "select count(*) from " + table;
+ }
+
+ public static void assertData(
+ String database, String table, int start, int end, BaseEnv baseEnv) {
+ TestUtils.assertDataEventuallyOnEnv(
+ baseEnv,
+ TableModelUtils.getQuerySql(table),
+ TableModelUtils.generateHeaderResults(),
+ TableModelUtils.generateExpectedResults(start, end),
+ database);
+ }
+
+ public static void assertData(String database, String table, Tablet tablet, BaseEnv baseEnv) {
+ TestUtils.assertDataEventuallyOnEnv(
+ baseEnv,
+ TableModelUtils.getQuerySql(table),
+ TableModelUtils.generateHeaderResults(),
+ TableModelUtils.generateExpectedResults(tablet),
+ database);
+ }
+
+ public static boolean hasDataBase(String database, BaseEnv baseEnv) {
+ TestUtils.assertDataEventuallyOnEnv(baseEnv, "", "", Collections.emptySet(), database);
+ return true;
+ }
+
+ public static void assertCountData(String database, String table, int count, BaseEnv baseEnv) {
+ TestUtils.assertDataEventuallyOnEnv(
+ baseEnv, getQueryCountSql(table), "_col0,", Collections.singleton(count + ","), database);
+ }
+
+ public static String getDateStr(int value) {
+ Date date = new Date(value);
+ SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd");
+ try {
+ return dateFormat.format(date);
+ } catch (Exception e) {
+ return "1970-01-01";
+ }
+ }
+
+ public static LocalDate getDate(int value) {
+ Date date = new Date(value);
+ SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd");
+ try {
+ return DateUtils.parseIntToLocalDate(
+ DateUtils.parseDateExpressionToInt(dateFormat.format(date)));
+ } catch (Exception e) {
+ return DateUtils.parseIntToLocalDate(DateUtils.parseDateExpressionToInt("1970-01-01"));
+ }
+ }
+
+ public static Tablet generateTablet(
+ String tableName, int start, int end, boolean allowNullValue) {
+ List<IMeasurementSchema> schemaList = new ArrayList<>();
+ schemaList.add(new MeasurementSchema("s0", TSDataType.STRING));
+ schemaList.add(new MeasurementSchema("s1", TSDataType.INT64));
+ schemaList.add(new MeasurementSchema("s2", TSDataType.FLOAT));
+ schemaList.add(new MeasurementSchema("s3", TSDataType.STRING));
+ schemaList.add(new MeasurementSchema("s4", TSDataType.TIMESTAMP));
+ schemaList.add(new MeasurementSchema("s5", TSDataType.INT32));
+ schemaList.add(new MeasurementSchema("s6", TSDataType.DOUBLE));
+ schemaList.add(new MeasurementSchema("s7", TSDataType.DATE));
+ schemaList.add(new MeasurementSchema("s8", TSDataType.TEXT));
+
+ final List<Tablet.ColumnType> columnTypes =
+ Arrays.asList(
+ Tablet.ColumnType.ID,
+ Tablet.ColumnType.MEASUREMENT,
+ Tablet.ColumnType.MEASUREMENT,
+ Tablet.ColumnType.MEASUREMENT,
+ Tablet.ColumnType.MEASUREMENT,
+ Tablet.ColumnType.MEASUREMENT,
+ Tablet.ColumnType.MEASUREMENT,
+ Tablet.ColumnType.MEASUREMENT,
+ Tablet.ColumnType.MEASUREMENT);
+ Tablet tablet = new Tablet(tableName, schemaList, columnTypes, end - start);
+ tablet.initBitMaps();
+ Random random = new Random();
+
+ // s2 float, s3 string, s4 timestamp, s5 int32, s6 double, s7 date, s8 text
+ for (long row = 0; row < end - start; row++) {
+ int randomNumber = allowNullValue ? random.nextInt(9) : 9;
+ long value = start + row;
+ int rowIndex = tablet.rowSize++;
+ tablet.addTimestamp(rowIndex, value);
+ tablet.addValue(
+ "s0", rowIndex, new Binary(String.valueOf(value).getBytes(StandardCharsets.UTF_8)));
+ tablet.addValue("s1", rowIndex, value);
+ tablet.addValue("s2", rowIndex, (value * 1.0f));
+ tablet.addValue(
+ "s3", rowIndex, new Binary(String.valueOf(value).getBytes(StandardCharsets.UTF_8)));
+ tablet.addValue("s4", rowIndex, value);
+ tablet.addValue("s5", rowIndex, (int) value);
+ tablet.addValue("s6", rowIndex, value * 0.1);
+ tablet.addValue("s7", rowIndex, getDate((int) value));
+ tablet.addValue(
+ "s8", rowIndex, new Binary(String.valueOf(value).getBytes(StandardCharsets.UTF_8)));
+ if (randomNumber < 9) {
+ tablet.addValue("s" + randomNumber, rowIndex, null);
+ }
+ }
+
+ return tablet;
+ }
+}