Pipe/Subscription: Add 'Create If Not Exists' and 'Drop If Exists' Support for Pipes, Plugins, and Topics

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

@@ -0,0 +1,228 @@
+/*
+ * 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.autocreate;
+
+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.it.env.cluster.node.DataNodeWrapper;
+import org.apache.iotdb.it.framework.IoTDBTestRunner;
+import org.apache.iotdb.itbase.category.MultiClusterIT2AutoCreateSchema;
+
+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({MultiClusterIT2AutoCreateSchema.class})
+public class IoTDBPipeConditionalOperationsIT extends AbstractPipeDualAutoIT {
+
+  @Test
+  public void testBasicCreatePipeIfNotExists() throws Exception {
+    final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+
+    // Create pipe
+    String sql =
+        String.format(
+            "create pipe If Not Exists a2b with source ('source'='iotdb-source', 'source.pattern'='root.test1', 'source.realtime.mode'='stream') with processor ('processor'='do-nothing-processor') with sink ('node-urls'='%s')",
+            receiverDataNode.getIpAndPortString());
+    try (final Connection connection = senderEnv.getConnection();
+        final Statement statement = connection.createStatement()) {
+      statement.execute(sql);
+    } catch (SQLException e) {
+      fail(e.getMessage());
+    }
+
+    // show pipe
+    long creationTime;
+    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("source.pattern=root.test1"));
+      Assert.assertTrue(
+          showPipeResult.get(0).pipeExtractor.contains("source.realtime.mode=stream"));
+      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
+      creationTime = showPipeResult.get(0).creationTime;
+    }
+
+    // Create pipe If Not Exists
+    sql =
+        String.format(
+            "create pipe If Not Exists a2b with source ('source'='iotdb-source', 'source.path'='root.test2.**') with sink ('node-urls'='%s')",
+            receiverDataNode.getIpAndPortString());
+    try (final Connection connection = senderEnv.getConnection();
+        final Statement statement = connection.createStatement()) {
+      statement.execute(sql);
+    } 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("source.pattern=root.test1"));
+      Assert.assertTrue(
+          showPipeResult.get(0).pipeExtractor.contains("source.realtime.mode=stream"));
+      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())));
+      Assert.assertFalse(showPipeResult.get(0).pipeExtractor.contains("source.path=root.test2.**"));
+      Assert.assertEquals(creationTime, showPipeResult.get(0).creationTime);
+    }
+  }
+
+  @Test
+  public void testBasicDropPipeIfExists() throws Exception {
+    final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+
+    // Create pipe
+    final String sql =
+        String.format(
+            "create pipe If Not Exists a2b with source ('source'='iotdb-source', 'source.path'='root.test1.**') with processor ('processor'='do-nothing-processor') with sink ('node-urls'='%s')",
+            receiverDataNode.getIpAndPortString());
+    try (final Connection connection = senderEnv.getConnection();
+        final Statement statement = connection.createStatement()) {
+      statement.execute(sql);
+    } catch (SQLException e) {
+      fail(e.getMessage());
+    }
+
+    // Drop pipe If Exists
+    try (final Connection connection = senderEnv.getConnection();
+        final Statement statement = connection.createStatement()) {
+      statement.execute("Drop pipe If Exists 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(0, showPipeResult.size());
+    }
+
+    // Drop pipe If Exists
+    try (final Connection connection = senderEnv.getConnection();
+        final Statement statement = connection.createStatement()) {
+      statement.execute("Drop pipe If Exists a2b");
+    } catch (SQLException e) {
+      fail(e.getMessage());
+    }
+  }
+
+  public void testBasicAlterPipeIfExists() throws Exception {
+    final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+
+    // Alter pipe If Exists
+    String sql =
+        String.format(
+            "Alter pipe If Exists a2b replace sink ('node-urls'='%s')",
+            receiverDataNode.getIpAndPortString());
+    try (final Connection connection = senderEnv.getConnection();
+        final Statement statement = connection.createStatement()) {
+      statement.execute(sql);
+    } 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(0, showPipeResult.size());
+    }
+
+    // Create pipe
+    sql =
+        String.format(
+            "create pipe If Not Exists a2b with source ('source'='iotdb-source', 'source.pattern'='root.test1', 'source.realtime.mode'='stream') with processor ('processor'='do-nothing-processor') with sink ('node-urls'='%s')",
+            receiverDataNode.getIpAndPortString());
+    try (final Connection connection = senderEnv.getConnection();
+        final Statement statement = connection.createStatement()) {
+      statement.execute(sql);
+    } catch (SQLException e) {
+      fail(e.getMessage());
+    }
+
+    // Alter pipe If Exists
+    sql =
+        String.format(
+            "Alter pipe If Exists a2b replace source () replace processor () replace sink ('node-urls'='%s')",
+            receiverDataNode.getIpAndPortString());
+    try (final Connection connection = senderEnv.getConnection();
+        final Statement statement = connection.createStatement()) {
+      statement.execute(sql);
+    } 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("source.pattern=root.test1"));
+      Assert.assertFalse(
+          showPipeResult.get(0).pipeExtractor.contains("source.realtime.mode=stream"));
+      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())));
+    }
+  }
+}

iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TSStatusCode.java

@@ -250,6 +250,7 @@ public enum TSStatusCode {
   ALTER_TOPIC_ERROR(2002),
   SHOW_TOPIC_ERROR(2003),
   TOPIC_PUSH_META_ERROR(2004),
+  TOPIC_NOT_EXIST_ERROR(2005),
 
   // Consumer
   CREATE_CONSUMER_ERROR(2100),

iotdb-core/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/IoTDBSqlParser.g4

@@ -535,7 +535,7 @@ verifyConnection
 
 // Pipe Task =========================================================================================
 createPipe
-    : CREATE PIPE pipeName=identifier
+    : CREATE PIPE  (IF NOT EXISTS)? pipeName=identifier
         extractorAttributesClause?
         processorAttributesClause?
         connectorAttributesClause
@@ -575,7 +575,7 @@ connectorAttributeClause
     ;
 
 alterPipe
-    : ALTER PIPE pipeName=identifier
+    : ALTER PIPE (IF EXISTS)? pipeName=identifier
         alterExtractorAttributesClause?
         alterProcessorAttributesClause?
         alterConnectorAttributesClause?
@@ -603,7 +603,7 @@ alterConnectorAttributesClause
     ;
 
 dropPipe
-    : DROP PIPE pipeName=identifier
+    : DROP PIPE (IF EXISTS)? pipeName=identifier
     ;
 
 startPipe
@@ -620,11 +620,11 @@ showPipes
 
 // Pipe Plugin =========================================================================================
 createPipePlugin
-    : CREATE PIPEPLUGIN pluginName=identifier AS className=STRING_LITERAL uriClause
+    : CREATE PIPEPLUGIN (IF NOT EXISTS)? pluginName=identifier AS className=STRING_LITERAL uriClause
     ;
 
 dropPipePlugin
-    : DROP PIPEPLUGIN pluginName=identifier
+    : DROP PIPEPLUGIN (IF EXISTS)? pluginName=identifier
     ;
 
 showPipePlugins
@@ -633,7 +633,7 @@ showPipePlugins
 
 // Topic =========================================================================================
 createTopic
-    : CREATE TOPIC topicName=identifier topicAttributesClause?
+    : CREATE TOPIC (IF NOT EXISTS)? topicName=identifier topicAttributesClause?
     ;
 
 topicAttributesClause
@@ -645,7 +645,7 @@ topicAttributeClause
     ;
 
 dropTopic
-    : DROP TOPIC topicName=identifier
+    : DROP TOPIC (IF EXISTS)? topicName=identifier
     ;
 
 showTopics

iotdb-core/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/SqlLexer.g4

@@ -286,6 +286,10 @@ EVERY
     : E V E R Y
     ;
 
+EXISTS
+    : E X I S T S
+    ;
+
 EXPLAIN
     : E X P L A I N
     ;
@@ -942,10 +946,15 @@ ELSE
     : E L S E
     ;
 
+IF
+    : I F
+    ;
+
 INF
     : I N F
     ;
 
+
 // Privileges Keywords
 
 PRIVILEGE_VALUE

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java

@@ -146,6 +146,9 @@
 import org.apache.iotdb.confignode.rpc.thrift.TDeleteLogicalViewReq;
 import org.apache.iotdb.confignode.rpc.thrift.TDeleteTimeSeriesReq;
 import org.apache.iotdb.confignode.rpc.thrift.TDropCQReq;
+import org.apache.iotdb.confignode.rpc.thrift.TDropPipePluginReq;
+import org.apache.iotdb.confignode.rpc.thrift.TDropPipeReq;
+import org.apache.iotdb.confignode.rpc.thrift.TDropTopicReq;
 import org.apache.iotdb.confignode.rpc.thrift.TDropTriggerReq;
 import org.apache.iotdb.confignode.rpc.thrift.TGetAllPipeInfoResp;
 import org.apache.iotdb.confignode.rpc.thrift.TGetAllSubscriptionInfoResp;
@@ -1441,10 +1444,10 @@ public TSStatus createPipePlugin(TCreatePipePluginReq req) {
   }
 
   @Override
-  public TSStatus dropPipePlugin(String pipePluginName) {
+  public TSStatus dropPipePlugin(TDropPipePluginReq req) {
     TSStatus status = confirmLeader();
     return status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()
-        ? pipeManager.getPipePluginCoordinator().dropPipePlugin(pipePluginName)
+        ? pipeManager.getPipePluginCoordinator().dropPipePlugin(req)
         : status;
   }
 
@@ -2026,10 +2029,10 @@ public TSStatus stopPipe(String pipeName) {
   }
 
   @Override
-  public TSStatus dropPipe(String pipeName) {
+  public TSStatus dropPipe(TDropPipeReq req) {
     TSStatus status = confirmLeader();
     return status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()
-        ? pipeManager.getPipeTaskCoordinator().dropPipe(pipeName)
+        ? pipeManager.getPipeTaskCoordinator().dropPipe(req)
         : status;
   }
 
@@ -2058,10 +2061,10 @@ public TSStatus createTopic(TCreateTopicReq req) {
   }
 
   @Override
-  public TSStatus dropTopic(String topicName) {
+  public TSStatus dropTopic(TDropTopicReq req) {
     TSStatus status = confirmLeader();
     return status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()
-        ? subscriptionManager.getSubscriptionCoordinator().dropTopic(topicName)
+        ? subscriptionManager.getSubscriptionCoordinator().dropTopic(req)
         : status;
   }