Support target database in write-back sink

integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/basic/IoTDBPipeDataSinkIT.java

@@ -277,10 +277,8 @@ public void testWriteBackSink() throws Exception {
       sourceAttributes.put("source.table-name", "test.*");
       sourceAttributes.put("user", "root");
 
-      processorAttributes.put("processor", "rename-database-processor");
-      processorAttributes.put("processor.new-db-name", "Test1");
-
       sinkAttributes.put("sink", "write-back-sink");
+      sinkAttributes.put("sink.database", "Test1");
       sinkAttributes.put("user", "root");
 
       final TSStatus status =

integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/manual/IoTDBPipeWriteBackSinkIT.java

@@ -0,0 +1,96 @@
+/*
+ * 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.dual.treemodel.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.framework.IoTDBTestRunner;
+import org.apache.iotdb.itbase.category.MultiClusterIT2DualTreeManual;
+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({MultiClusterIT2DualTreeManual.class})
+public class IoTDBPipeWriteBackSinkIT extends AbstractPipeDualTreeModelManualIT {
+
+  @Test
+  public void testWriteBackSinkWithTargetDatabaseForTreeModel() throws Exception {
+    TestUtils.executeNonQueries(
+        senderEnv,
+        Arrays.asList(
+            "create database root.source",
+            "create timeseries root.source.d1.s1 with datatype=INT32,encoding=PLAIN",
+            "create database root.target.db",
+            "create timeseries root.target.db.d1.s1 with datatype=INT32,encoding=PLAIN"),
+        null);
+
+    try (final SyncConfigNodeIServiceClient client =
+        (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+      final Map<String, String> sourceAttributes = new HashMap<>();
+      final Map<String, String> processorAttributes = new HashMap<>();
+      final Map<String, String> sinkAttributes = new HashMap<>();
+
+      sourceAttributes.put("source.inclusion", "data.insert");
+      sourceAttributes.put("source.forwarding-pipe-requests", "false");
+      sourceAttributes.put("source.path", "root.source.**");
+      sourceAttributes.put("user", "root");
+
+      sinkAttributes.put("sink", "write-back-sink");
+      sinkAttributes.put("sink.database", "root.target.db");
+      sinkAttributes.put("user", "root");
+
+      final TSStatus status =
+          client.createPipe(
+              new TCreatePipeReq("testPipe", sinkAttributes)
+                  .setExtractorAttributes(sourceAttributes)
+                  .setProcessorAttributes(processorAttributes));
+      Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+
+      Assert.assertEquals(
+          TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("testPipe").getCode());
+    }
+
+    TestUtils.executeNonQueries(
+        senderEnv,
+        Arrays.asList(
+            "insert into root.source.d1(time, s1) values (1, 1)",
+            "insert into root.source.d1(time, s1) values (2, 2)",
+            "flush"),
+        null);
+
+    TestUtils.assertDataEventuallyOnEnv(
+        senderEnv,
+        "select * from root.target.db.**",
+        "Time,root.target.db.d1.s1,",
+        Collections.unmodifiableSet(new HashSet<>(Arrays.asList("1,1,", "2,2,"))));
+  }
+}

iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java

@@ -1449,6 +1449,21 @@ public final class DataNodePipeMessages {
   // ---------------------------------------------------------------------------
   public static final String PLUGIN_NOT_REGISTERED_FMT = "plugin %s is not registered.";
 
+  // ---------------------------------------------------------------------------
+  // pipe - WriteBackSink
+  // ---------------------------------------------------------------------------
+  public static final String TABLE_MODEL_DATABASE_INVALID_FMT =
+      "The table-model database %s is invalid. It should not contain '%s', should match the "
+          + "pattern %s, and the length should not exceed %d";
+  public static final String TREE_MODEL_DATABASE_INVALID_FMT =
+      "The tree-model database %s is invalid. It should be a legal tree-model database path, "
+          + "should match the pattern %s, and the length should not exceed %d";
+  public static final String TARGET_TREE_MODEL_DATABASE_CANNOT_BE_USED_FOR_TABLE_MODEL_EVENTS_FMT =
+      "The target tree-model database %s cannot be used for table-model events because the "
+          + "corresponding table-model database %s is invalid.";
+  public static final String FAILED_TO_REWRITE_TREE_MODEL_DATABASE_FMT =
+      "Failed to rewrite tree-model database from %s to %s for device %s.";
+
   // ---------------------------------------------------------------------------
   // pipe – PipeTransferTrackableHandler
   // ---------------------------------------------------------------------------

iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java

@@ -1396,6 +1396,18 @@ public final class DataNodePipeMessages {
   // ---------------------------------------------------------------------------
   public static final String PLUGIN_NOT_REGISTERED_FMT = "插件 %s 未注册。";
 
+  // ---------------------------------------------------------------------------
+  // pipe - WriteBackSink
+  // ---------------------------------------------------------------------------
+  public static final String TABLE_MODEL_DATABASE_INVALID_FMT =
+      "表模型数据库 %s 非法：不应包含 '%s'，应匹配 %s，且长度不应超过 %d";
+  public static final String TREE_MODEL_DATABASE_INVALID_FMT =
+      "树模型数据库 %s 非法：应为合法的树模型数据库路径，应匹配 %s，且长度不应超过 %d";
+  public static final String TARGET_TREE_MODEL_DATABASE_CANNOT_BE_USED_FOR_TABLE_MODEL_EVENTS_FMT =
+      "目标树模型数据库 %s 不能用于表模型事件，因为对应的表模型数据库 %s 非法。";
+  public static final String FAILED_TO_REWRITE_TREE_MODEL_DATABASE_FMT =
+      "将树模型数据库从 %s 重写为 %s 失败，设备为 %s。";
+
   // ---------------------------------------------------------------------------
   // pipe – PipeTransferTrackableHandler
   // ---------------------------------------------------------------------------