Pipe: Fixed the bug that historical aligned timeseries' attributes/tags/alias is not transferred (#14071) (#14078)

Author: Caideyipi

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

@@ -261,4 +261,60 @@ public void testAuthInclusion() throws Exception {
           new HashSet<>(Arrays.asList("admin,", "test,")));
     }
   }
+
+  @Test
+  public void testTimeSeriesInclusion() 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()) {
+
+      // 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(
+              "create database root.sg",
+              "create timeseries root.sg.a.b int32",
+              "create aligned timeseries root.sg.`apache|timecho-tag-attr`.d1(s1 INT32 tags(tag1=v1, tag2=v2) attributes(attr1=v1, attr2=v2), s2 DOUBLE tags(tag3=v3, tag4=v4) attributes(attr3=v3, attr4=v4))"))) {
+        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.inclusion", "schema");
+
+      connectorAttributes.put("connector", "iotdb-thrift-connector");
+      connectorAttributes.put("connector.ip", receiverIp);
+      connectorAttributes.put("connector.port", Integer.toString(receiverPort));
+      connectorAttributes.put("connector.exception.conflict.resolve-strategy", "retry");
+      connectorAttributes.put("connector.exception.conflict.retry-max-time-seconds", "-1");
+
+      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());
+
+      TestUtils.assertDataEventuallyOnEnv(
+          receiverEnv,
+          "show timeseries",
+          "Timeseries,Alias,Database,DataType,Encoding,Compression,Tags,Attributes,Deadband,DeadbandParameters,ViewType,",
+          new HashSet<>(
+              Arrays.asList(
+                  "root.sg.a.b,null,root.sg,INT32,TS_2DIFF,LZ4,null,null,null,null,BASE,",
+                  "root.sg.`apache|timecho-tag-attr`.d1.s1,null,root.sg,INT32,TS_2DIFF,LZ4,{\"tag1\":\"v1\",\"tag2\":\"v2\"},{\"attr2\":\"v2\",\"attr1\":\"v1\"},null,null,BASE,",
+                  "root.sg.`apache|timecho-tag-attr`.d1.s2,null,root.sg,DOUBLE,GORILLA,LZ4,{\"tag4\":\"v4\",\"tag3\":\"v3\"},{\"attr4\":\"v4\",\"attr3\":\"v3\"},null,null,BASE,")));
+    }
+  }
 }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/schemaregion/SchemaExecutionVisitor.java

@@ -79,6 +79,7 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 
 /** Schema write {@link PlanNode} visitor */
 public class SchemaExecutionVisitor extends PlanVisitor<TSStatus, ISchemaRegion> {
@@ -277,16 +278,21 @@ private void executeInternalCreateAlignedTimeSeries(
     final List<TSDataType> dataTypeList = measurementGroup.getDataTypes();
     final List<TSEncoding> encodingList = measurementGroup.getEncodings();
     final List<CompressionType> compressionTypeList = measurementGroup.getCompressors();
+    final List<String> aliasList = measurementGroup.getAliasList();
+    final List<Map<String, String>> tagsList = measurementGroup.getTagsList();
+    final List<Map<String, String>> attributesList = measurementGroup.getAttributesList();
+
     final ICreateAlignedTimeSeriesPlan createAlignedTimeSeriesPlan =
         SchemaRegionWritePlanFactory.getCreateAlignedTimeSeriesPlan(
             devicePath,
             measurementList,
             dataTypeList,
             encodingList,
             compressionTypeList,
-            null,
-            null,
-            null);
+            aliasList,
+            tagsList,
+            attributesList);
+
     // With merge is only true for pipe to upsert the receiver alias/tags/attributes in historical
     // transfer.
     // For normal internal creation, the alias/tags/attributes are not set
@@ -313,6 +319,16 @@ private void executeInternalCreateAlignedTimeSeries(
         encodingList.remove(index);
         compressionTypeList.remove(index);
 
+        if (Objects.nonNull(aliasList)) {
+          aliasList.remove(index);
+        }
+        if (Objects.nonNull(tagsList)) {
+          tagsList.remove(index);
+        }
+        if (Objects.nonNull(attributesList)) {
+          attributesList.remove(index);
+        }
+
         // If with merge is set, the lists are deep copied and need to be altered here.
         // We still remove the element from the original list to help cascading pipe transfer
         // schema.
@@ -323,6 +339,16 @@ private void executeInternalCreateAlignedTimeSeries(
           createAlignedTimeSeriesPlan.getDataTypes().remove(index);
           createAlignedTimeSeriesPlan.getEncodings().remove(index);
           createAlignedTimeSeriesPlan.getCompressors().remove(index);
+
+          if (Objects.nonNull(aliasList)) {
+            createAlignedTimeSeriesPlan.getAliasList().remove(index);
+          }
+          if (Objects.nonNull(tagsList)) {
+            createAlignedTimeSeriesPlan.getTagsList().remove(index);
+          }
+          if (Objects.nonNull(attributesList)) {
+            createAlignedTimeSeriesPlan.getAttributesList().remove(index);
+          }
         }
 
         if (measurementList.isEmpty()) {