Pipe: Implement table meta sync on configNode

integration-test/src/test/java/org/apache/iotdb/db/it/utils/TestUtils.java

@@ -48,6 +48,7 @@
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Set;
 import java.util.TreeMap;
 import java.util.concurrent.TimeUnit;
@@ -1037,8 +1038,8 @@ public static void assertDataEventuallyOnEnv(
                     TestUtils.assertResultSetEqual(
                         executeQueryWithRetry(statement, sql), expectedHeader, expectedResSet);
                   }
-                } catch (Exception e) {
-                  Assert.fail();
+                } catch (final Exception e) {
+                  Assert.fail(e.getMessage());
                 }
               });
     } catch (Exception e) {
@@ -1077,17 +1078,32 @@ public static void assertDataEventuallyOnEnv(
   }
 
   public static void assertDataAlwaysOnEnv(
-      BaseEnv env, String sql, String expectedHeader, Set<String> expectedResSet) {
-    assertDataAlwaysOnEnv(env, sql, expectedHeader, expectedResSet, 10);
+      final BaseEnv env,
+      final String sql,
+      final String expectedHeader,
+      final Set<String> expectedResSet) {
+    assertDataAlwaysOnEnv(env, sql, expectedHeader, expectedResSet, 10, null);
+  }
+
+  public static void assertDataAlwaysOnEnv(
+      final BaseEnv env,
+      final String sql,
+      final String expectedHeader,
+      final Set<String> expectedResSet,
+      final String database) {
+    assertDataAlwaysOnEnv(env, sql, expectedHeader, expectedResSet, 10, database);
   }
 
   public static void assertDataAlwaysOnEnv(
       BaseEnv env,
       String sql,
       String expectedHeader,
       Set<String> expectedResSet,
-      long consistentSeconds) {
-    try (Connection connection = env.getConnection();
+      long consistentSeconds,
+      final String database) {
+    try (Connection connection =
+            env.getConnection(
+                Objects.isNull(database) ? BaseEnv.TREE_SQL_DIALECT : BaseEnv.TABLE_SQL_DIALECT);
         Statement statement = connection.createStatement()) {
       // Keep retrying if there are execution failures
       await()
@@ -1098,6 +1114,9 @@ public static void assertDataAlwaysOnEnv(
           .failFast(
               () -> {
                 try {
+                  if (Objects.nonNull(database)) {
+                    statement.execute("use " + database);
+                  }
                   TestUtils.assertResultSetEqual(
                       executeQueryWithRetry(statement, sql), expectedHeader, expectedResSet);
                 } catch (Exception e) {

integration-test/src/test/java/org/apache/iotdb/pipe/it/manual/IoTDBPipeTableManualIT.java

@@ -0,0 +1,263 @@
+/*
+ * 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.manual;
+
+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.MultiClusterIT2ManualCreateSchema;
+import org.apache.iotdb.itbase.env.BaseEnv;
+import org.apache.iotdb.pipe.it.tablemodel.TableModelUtils;
+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.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+
+@RunWith(IoTDBTestRunner.class)
+@Category({MultiClusterIT2ManualCreateSchema.class})
+public class IoTDBPipeTableManualIT extends AbstractPipeDualManualIT {
+  @Test
+  public void testTableSync() 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.inclusion", "all");
+      extractorAttributes.put("extractor.inclusion.exclusion", "data.delete");
+      extractorAttributes.put("extractor.capture.tree", "false");
+      extractorAttributes.put("extractor.capture.table", "true");
+      extractorAttributes.put("extractor.database-name", "test");
+      extractorAttributes.put("extractor.table-name", "t.*[0-9]");
+
+      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("testPipe", connectorAttributes)
+                  .setExtractorAttributes(extractorAttributes)
+                  .setProcessorAttributes(processorAttributes));
+
+      Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+
+      Assert.assertEquals(
+          TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("testPipe").getCode());
+
+      final String dbName = "test";
+      TableModelUtils.createDataBase(senderEnv, dbName, 300);
+
+      if (!TestUtils.tryExecuteNonQueriesWithRetry(
+          dbName,
+          BaseEnv.TABLE_SQL_DIALECT,
+          senderEnv,
+          Arrays.asList(
+              "create table table1(a id, b attribute, c int32) with (ttl=3000)",
+              "alter table table1 add column d int64",
+              "alter table table1 drop column b",
+              "alter table table1 set properties ttl=default",
+              "insert into table1 (a, c, d) values(1, 1, 1)",
+              "delete devices from table1",
+              "create table noTransferTable(a id, b attribute, c int32) with (ttl=3000)"))) {
+        return;
+      }
+
+      TableModelUtils.createDataBase(senderEnv, "noTransferDatabase", 300);
+
+      TestUtils.assertDataEventuallyOnEnv(
+          receiverEnv,
+          "show tables from test",
+          "TableName,TTL(ms),",
+          Collections.singleton("table1,300,"),
+          dbName);
+
+      // Will not include no-transfer table
+      TestUtils.assertDataAlwaysOnEnv(
+          receiverEnv,
+          "show tables from test",
+          "TableName,TTL(ms),",
+          Collections.singleton("table1,300,"),
+          dbName);
+
+      TestUtils.assertDataEventuallyOnEnv(
+          receiverEnv,
+          "desc table1",
+          "ColumnName,DataType,Category,",
+          new HashSet<>(
+              Arrays.asList(
+                  "time,TIMESTAMP,TIME,",
+                  "a,STRING,ID,",
+                  "c,INT32,MEASUREMENT,",
+                  "d,INT64,MEASUREMENT,")),
+          dbName);
+
+      if (!TestUtils.tryExecuteNonQueryWithRetry(
+          dbName, BaseEnv.TABLE_SQL_DIALECT, senderEnv, "drop table table1")) {
+        return;
+      }
+
+      TestUtils.assertDataEventuallyOnEnv(
+          receiverEnv,
+          "show tables from test",
+          "TableName,TTL(ms),",
+          Collections.emptySet(),
+          dbName);
+
+      if (!TestUtils.tryExecuteNonQueryWithRetry(
+          dbName, BaseEnv.TABLE_SQL_DIALECT, senderEnv, "drop database test")) {
+        return;
+      }
+
+      // Will not include no-transfer database
+      TestUtils.assertDataEventuallyOnEnv(
+          receiverEnv,
+          "show databases",
+          "Database,TTL(ms),SchemaReplicationFactor,DataReplicationFactor,TimePartitionInterval,",
+          Collections.emptySet(),
+          null);
+    }
+  }
+
+  @Test
+  public void testNoTree() 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.inclusion", "all");
+      extractorAttributes.put("extractor.inclusion.exclusion", "data.delete");
+      extractorAttributes.put("extractor.capture.tree", "false");
+      extractorAttributes.put("extractor.capture.table", "true");
+
+      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("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(
+              "create database root.test",
+              "alter database root.test with schema_region_group_num=2, data_region_group_num=3",
+              "create timeSeries root.test.d1.s1 int32",
+              "insert into root.test.d1 (s1) values (1)"))) {
+        return;
+      }
+
+      TestUtils.assertDataAlwaysOnEnv(
+          receiverEnv,
+          "show databases",
+          "Database,SchemaReplicationFactor,DataReplicationFactor,TimePartitionInterval,",
+          Collections.emptySet());
+    }
+  }
+
+  @Test
+  public void testNoTable() 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.inclusion", "all");
+      extractorAttributes.put("extractor.inclusion.exclusion", "data.delete");
+      extractorAttributes.put("extractor.capture.tree", "true");
+      extractorAttributes.put("extractor.capture.table", "false");
+
+      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("testPipe", connectorAttributes)
+                  .setExtractorAttributes(extractorAttributes)
+                  .setProcessorAttributes(processorAttributes));
+
+      Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+
+      Assert.assertEquals(
+          TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("testPipe").getCode());
+
+      final String dbName = "test";
+      TableModelUtils.createDataBase(senderEnv, dbName, 300);
+
+      if (!TestUtils.tryExecuteNonQueriesWithRetry(
+          dbName,
+          BaseEnv.TABLE_SQL_DIALECT,
+          senderEnv,
+          Arrays.asList(
+              "create table table1(a id, b attribute, c int32) with (ttl=3000)",
+              "alter table table1 add column d int64",
+              "alter table table1 drop column b",
+              "alter table table1 set properties ttl=default"))) {
+        return;
+      }
+
+      TestUtils.assertDataAlwaysOnEnv(
+          receiverEnv,
+          "show databases",
+          "Database,TTL(ms),SchemaReplicationFactor,DataReplicationFactor,TimePartitionInterval,",
+          Collections.emptySet(),
+          dbName);
+    }
+  }
+}

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

@@ -77,10 +77,10 @@ public static void createDataBaseAndTable(BaseEnv baseEnv, String table, String
     }
   }
 
-  public static void createDataBase(BaseEnv baseEnv, String database) {
+  public static void createDataBase(BaseEnv baseEnv, String database, long ttl) {
     try (Connection connection = baseEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT);
         Statement statement = connection.createStatement()) {
-      statement.execute("create database if not exists " + database);
+      statement.execute("create database if not exists " + database + " with (ttl=" + ttl + ")");
     } catch (Exception e) {
       fail(e.getMessage());
     }

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlan.java

@@ -58,7 +58,9 @@
 import org.apache.iotdb.confignode.consensus.request.write.partition.CreateSchemaPartitionPlan;
 import org.apache.iotdb.confignode.consensus.request.write.partition.RemoveRegionLocationPlan;
 import org.apache.iotdb.confignode.consensus.request.write.partition.UpdateRegionLocationPlan;
+import org.apache.iotdb.confignode.consensus.request.write.pipe.payload.PipeCreateTablePlan;
 import org.apache.iotdb.confignode.consensus.request.write.pipe.payload.PipeDeactivateTemplatePlan;
+import org.apache.iotdb.confignode.consensus.request.write.pipe.payload.PipeDeleteDevicesPlan;
 import org.apache.iotdb.confignode.consensus.request.write.pipe.payload.PipeDeleteLogicalViewPlan;
 import org.apache.iotdb.confignode.consensus.request.write.pipe.payload.PipeDeleteTimeSeriesPlan;
 import org.apache.iotdb.confignode.consensus.request.write.pipe.payload.PipeEnrichedPlan;
@@ -200,6 +202,9 @@ public static ConfigPhysicalPlan create(final ByteBuffer buffer) throws IOExcept
         case AlterDatabase:
           plan = new DatabaseSchemaPlan(ConfigPhysicalPlanType.AlterDatabase);
           break;
+        case DeleteDatabaseV2:
+          plan = new DatabaseSchemaPlan(ConfigPhysicalPlanType.DeleteDatabaseV2);
+          break;
         case SetTTL:
           plan = new SetTTLPlan();
           break;
@@ -334,7 +339,7 @@ public static ConfigPhysicalPlan create(final ByteBuffer buffer) throws IOExcept
           plan = new ExtendSchemaTemplatePlan();
           break;
         case PreCreateTable:
-          plan = new PreCreateTablePlan();
+          plan = new PreCreateTablePlan(configPhysicalPlanType);
           break;
         case RollbackCreateTable:
           plan = new RollbackCreateTablePlan();
@@ -444,6 +449,12 @@ public static ConfigPhysicalPlan create(final ByteBuffer buffer) throws IOExcept
         case PipeDeactivateTemplate:
           plan = new PipeDeactivateTemplatePlan();
           break;
+        case PipeCreateTable:
+          plan = new PipeCreateTablePlan();
+          break;
+        case PipeDeleteDevices:
+          plan = new PipeDeleteDevicesPlan();
+          break;
         case UpdateTriggersOnTransferNodes:
           plan = new UpdateTriggersOnTransferNodesPlan();
           break;