Pipe: tree/table model isolation for alter/start/stop/drop/show pipe operations & support 'mode.double-living'

integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeAlterIT.java

@@ -26,7 +26,6 @@
 import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper;
 import org.apache.iotdb.it.framework.IoTDBTestRunner;
 import org.apache.iotdb.itbase.category.MultiClusterIT2AutoCreateSchema;
-import org.apache.iotdb.itbase.env.BaseEnv;
 
 import org.junit.Assert;
 import org.junit.Test;
@@ -106,7 +105,7 @@ public void testBasicAlterPipe() throws Exception {
     }
 
     // Alter pipe (modify)
-    try (final Connection connection = senderEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT);
+    try (final Connection connection = senderEnv.getConnection();
         final Statement statement = connection.createStatement()) {
       statement.execute("alter pipe a2b modify source ('source.pattern'='root.test2')");
     } catch (SQLException e) {

integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeSwitchStatusIT.java

@@ -266,11 +266,14 @@ public void testWrongPipeName() throws Exception {
                   .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());
+          TSStatusCode.PIPE_NOT_EXIST_ERROR.getStatusCode(), client.startPipe("").getCode());
+      Assert.assertEquals(
+          TSStatusCode.PIPE_NOT_EXIST_ERROR.getStatusCode(), client.startPipe("p0").getCode());
+      Assert.assertEquals(
+          TSStatusCode.PIPE_NOT_EXIST_ERROR.getStatusCode(), client.startPipe("p").getCode());
+      Assert.assertEquals(
+          TSStatusCode.PIPE_NOT_EXIST_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")));
@@ -281,10 +284,14 @@ public void testWrongPipeName() throws Exception {
       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());
+      Assert.assertEquals(
+          TSStatusCode.PIPE_NOT_EXIST_ERROR.getStatusCode(), client.stopPipe("").getCode());
+      Assert.assertEquals(
+          TSStatusCode.PIPE_NOT_EXIST_ERROR.getStatusCode(), client.stopPipe("p0").getCode());
+      Assert.assertEquals(
+          TSStatusCode.PIPE_NOT_EXIST_ERROR.getStatusCode(), client.stopPipe("p").getCode());
+      Assert.assertEquals(
+          TSStatusCode.PIPE_NOT_EXIST_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")));

integration-test/src/test/java/org/apache/iotdb/pipe/it/tablemodel/IoTDBPipeIsolationIT.java

@@ -0,0 +1,316 @@
+/*
+ * 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.TAlterPipeReq;
+import org.apache.iotdb.confignode.rpc.thrift.TCreatePipeReq;
+import org.apache.iotdb.confignode.rpc.thrift.TDropPipeReq;
+import org.apache.iotdb.confignode.rpc.thrift.TStartPipeReq;
+import org.apache.iotdb.confignode.rpc.thrift.TStopPipeReq;
+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.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+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 IoTDBPipeIsolationIT extends AbstractPipeTableModelTestIT {
+
+  @Test
+  public void testWritePipeIsolation() throws Exception {
+    final String treePipeName = "tree_a2b";
+    final String tablePipeName = "table_a2b";
+
+    final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+
+    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("__system.sql-dialect", "tree");
+      connectorAttributes.put("node-urls", receiverDataNode.getIpAndPortString());
+
+      final TSStatus status =
+          client.createPipe(
+              new TCreatePipeReq(treePipeName, connectorAttributes)
+                  .setExtractorAttributes(extractorAttributes)
+                  .setProcessorAttributes(processorAttributes));
+      Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+    }
+
+    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("__system.sql-dialect", "table");
+      connectorAttributes.put("node-urls", receiverDataNode.getIpAndPortString());
+
+      final TSStatus status =
+          client.createPipe(
+              new TCreatePipeReq(tablePipeName, connectorAttributes)
+                  .setExtractorAttributes(extractorAttributes)
+                  .setProcessorAttributes(processorAttributes));
+      Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+    }
+
+    try (final SyncConfigNodeIServiceClient client =
+        (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+      // start pipe
+      Assert.assertEquals(
+          TSStatusCode.PIPE_NOT_EXIST_ERROR.getStatusCode(),
+          client
+              .startPipeExtended(
+                  new TStartPipeReq(treePipeName).setSqlDialect(BaseEnv.TABLE_SQL_DIALECT))
+              .getCode());
+      Assert.assertEquals(
+          TSStatusCode.PIPE_NOT_EXIST_ERROR.getStatusCode(),
+          client
+              .startPipeExtended(
+                  new TStartPipeReq(tablePipeName).setSqlDialect(BaseEnv.TREE_SQL_DIALECT))
+              .getCode());
+      Assert.assertEquals(
+          TSStatusCode.SUCCESS_STATUS.getStatusCode(),
+          client
+              .startPipeExtended(
+                  new TStartPipeReq(treePipeName).setSqlDialect(BaseEnv.TREE_SQL_DIALECT))
+              .getCode());
+      Assert.assertEquals(
+          TSStatusCode.SUCCESS_STATUS.getStatusCode(),
+          client
+              .startPipeExtended(
+                  new TStartPipeReq(tablePipeName).setSqlDialect(BaseEnv.TABLE_SQL_DIALECT))
+              .getCode());
+
+      // stop pipe
+      Assert.assertEquals(
+          TSStatusCode.PIPE_NOT_EXIST_ERROR.getStatusCode(),
+          client
+              .stopPipeExtended(
+                  new TStopPipeReq(treePipeName).setSqlDialect(BaseEnv.TABLE_SQL_DIALECT))
+              .getCode());
+      Assert.assertEquals(
+          TSStatusCode.PIPE_NOT_EXIST_ERROR.getStatusCode(),
+          client
+              .stopPipeExtended(
+                  new TStopPipeReq(tablePipeName).setSqlDialect(BaseEnv.TREE_SQL_DIALECT))
+              .getCode());
+      Assert.assertEquals(
+          TSStatusCode.SUCCESS_STATUS.getStatusCode(),
+          client
+              .stopPipeExtended(
+                  new TStopPipeReq(treePipeName).setSqlDialect(BaseEnv.TREE_SQL_DIALECT))
+              .getCode());
+      Assert.assertEquals(
+          TSStatusCode.SUCCESS_STATUS.getStatusCode(),
+          client
+              .stopPipeExtended(
+                  new TStopPipeReq(tablePipeName).setSqlDialect(BaseEnv.TABLE_SQL_DIALECT))
+              .getCode());
+
+      // alter pipe
+      Assert.assertEquals(
+          TSStatusCode.PIPE_NOT_EXIST_ERROR.getStatusCode(),
+          client
+              .alterPipe(
+                  new TAlterPipeReq(
+                          treePipeName,
+                          Collections.emptyMap(),
+                          Collections.emptyMap(),
+                          false,
+                          false)
+                      .setExtractorAttributes(Collections.emptyMap())
+                      .setIsReplaceAllExtractorAttributes(false)
+                      .setSqlDialect(BaseEnv.TABLE_SQL_DIALECT))
+              .getCode());
+      Assert.assertEquals(
+          TSStatusCode.PIPE_NOT_EXIST_ERROR.getStatusCode(),
+          client
+              .alterPipe(
+                  new TAlterPipeReq(
+                          tablePipeName,
+                          Collections.emptyMap(),
+                          Collections.emptyMap(),
+                          false,
+                          false)
+                      .setExtractorAttributes(Collections.emptyMap())
+                      .setIsReplaceAllExtractorAttributes(false)
+                      .setSqlDialect(BaseEnv.TREE_SQL_DIALECT))
+              .getCode());
+      Assert.assertEquals(
+          TSStatusCode.SUCCESS_STATUS.getStatusCode(),
+          client
+              .alterPipe(
+                  new TAlterPipeReq(
+                          treePipeName,
+                          Collections.emptyMap(),
+                          Collections.emptyMap(),
+                          false,
+                          false)
+                      .setExtractorAttributes(Collections.emptyMap())
+                      .setIsReplaceAllExtractorAttributes(false)
+                      .setSqlDialect(BaseEnv.TREE_SQL_DIALECT))
+              .getCode());
+      Assert.assertEquals(
+          TSStatusCode.SUCCESS_STATUS.getStatusCode(),
+          client
+              .alterPipe(
+                  new TAlterPipeReq(
+                          tablePipeName,
+                          Collections.emptyMap(),
+                          Collections.emptyMap(),
+                          false,
+                          false)
+                      .setExtractorAttributes(Collections.emptyMap())
+                      .setIsReplaceAllExtractorAttributes(false)
+                      .setSqlDialect(BaseEnv.TABLE_SQL_DIALECT))
+              .getCode());
+
+      // drop pipe
+      Assert.assertEquals(
+          TSStatusCode.PIPE_NOT_EXIST_ERROR.getStatusCode(),
+          client
+              .dropPipeExtended(
+                  new TDropPipeReq(treePipeName).setSqlDialect(BaseEnv.TABLE_SQL_DIALECT))
+              .getCode());
+      Assert.assertEquals(
+          TSStatusCode.PIPE_NOT_EXIST_ERROR.getStatusCode(),
+          client
+              .dropPipeExtended(
+                  new TDropPipeReq(tablePipeName).setSqlDialect(BaseEnv.TREE_SQL_DIALECT))
+              .getCode());
+      Assert.assertEquals(
+          TSStatusCode.SUCCESS_STATUS.getStatusCode(),
+          client
+              .dropPipeExtended(
+                  new TDropPipeReq(treePipeName).setSqlDialect(BaseEnv.TREE_SQL_DIALECT))
+              .getCode());
+      Assert.assertEquals(
+          TSStatusCode.SUCCESS_STATUS.getStatusCode(),
+          client
+              .dropPipeExtended(
+                  new TDropPipeReq(tablePipeName).setSqlDialect(BaseEnv.TABLE_SQL_DIALECT))
+              .getCode());
+    }
+  }
+
+  @Test
+  public void testReadPipeIsolation() {
+    final String treePipeName = "tree_a2b";
+    final String tablePipeName = "table_a2b";
+
+    final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+
+    // Create tree pipe by tree session
+    try (final Connection connection = senderEnv.getConnection(BaseEnv.TREE_SQL_DIALECT);
+        final Statement statement = connection.createStatement()) {
+      statement.execute(
+          String.format(
+              "create pipe %s with sink ('node-urls'='%s')",
+              treePipeName, receiverDataNode.getIpAndPortString()));
+    } catch (SQLException e) {
+      fail(e.getMessage());
+    }
+
+    // Show tree pipe by tree session
+    try (final Connection connection = senderEnv.getConnection(BaseEnv.TREE_SQL_DIALECT);
+        final Statement statement = connection.createStatement()) {
+      final ResultSet resultSet = statement.executeQuery("show pipes");
+      int count = 0;
+      while (resultSet.next()) {
+        count++;
+      }
+      Assert.assertEquals(1, count);
+    } catch (SQLException e) {
+      fail(e.getMessage());
+    }
+
+    // Show table pipe by table session
+    try (final Connection connection = senderEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT);
+        final Statement statement = connection.createStatement()) {
+      final ResultSet resultSet = statement.executeQuery("show pipes");
+      int count = 0;
+      while (resultSet.next()) {
+        count++;
+      }
+      Assert.assertEquals(0, count);
+    } catch (SQLException e) {
+      fail(e.getMessage());
+    }
+
+    // Create table pipe by table session
+    try (final Connection connection = senderEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT);
+        final Statement statement = connection.createStatement()) {
+      statement.execute(
+          String.format(
+              "create pipe %s with sink ('node-urls'='%s')",
+              tablePipeName, receiverDataNode.getIpAndPortString()));
+    } catch (SQLException e) {
+      fail(e.getMessage());
+    }
+
+    // Show tree pipe by tree session
+    try (final Connection connection = senderEnv.getConnection(BaseEnv.TREE_SQL_DIALECT);
+        final Statement statement = connection.createStatement()) {
+      final ResultSet resultSet = statement.executeQuery("show pipes");
+      int count = 0;
+      while (resultSet.next()) {
+        count++;
+      }
+      Assert.assertEquals(1, count);
+    } catch (SQLException e) {
+      fail(e.getMessage());
+    }
+
+    // Show table pipe by table session
+    try (final Connection connection = senderEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT);
+        final Statement statement = connection.createStatement()) {
+      final ResultSet resultSet = statement.executeQuery("show pipes");
+      int count = 0;
+      while (resultSet.next()) {
+        count++;
+      }
+      Assert.assertEquals(1, count);
+    } catch (SQLException e) {
+      fail(e.getMessage());
+    }
+  }
+}