Pipe: Fixed the bug that attribute sync may cause ArrayIndexOutOfBoundsException / eliminate the existing attributes at the receiver which == null at the sender (#17539)

Author: Caideyipi

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

@@ -1055,16 +1055,15 @@ public static void executeNonQueries(BaseEnv env, List<String> sqlList) {
         null);
   }
 
-  public static void executeNonQueries(
-      BaseEnv env, List<String> sqlList, Connection defaultConnection) {
+  public static void executeNonQueries(BaseEnv env, List<String> sqlList, String sqlDialect) {
     executeNonQueries(
         env,
         sqlList,
         SessionConfig.DEFAULT_USER,
         SessionConfig.DEFAULT_PASSWORD,
         null,
-        TREE_SQL_DIALECT,
-        defaultConnection);
+        sqlDialect,
+        null);
   }
 
   public static void executeNonQueries(

integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/enhanced/IoTDBPipeMetaIT.java

@@ -232,8 +232,7 @@ public void testNoTree() throws Exception {
               "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)"),
-          null);
+              "insert into root.test.d1 (s1) values (1)"));
 
       TestUtils.assertDataAlwaysOnEnv(
           receiverEnv,
@@ -418,4 +417,35 @@ public void testValidation() throws Exception {
       }
     }
   }
+
+  @Test
+  public void testAttributeSync() {
+    TestUtils.executeNonQueries(
+        receiverEnv,
+        Arrays.asList(
+            "create database test",
+            "use test",
+            "create table table1(a tag, b attribute, c attribute, d int32)",
+            "insert into table1 (time, a, b, c, d) values(1, 1, null, 1, 1), (2, 2, 2, null, 2)"),
+        BaseEnv.TABLE_SQL_DIALECT);
+
+    TestUtils.executeNonQueries(
+        senderEnv,
+        Arrays.asList(
+            "create database test",
+            "use test",
+            "create table table1(a tag, b attribute, c attribute, d int32)",
+            "insert into table1 (time, a, b, c, d) values(1, 1, 1, null, 1), (2, 2, null, 2, 2)",
+            String.format(
+                "create pipe a2b with source ('inclusion'='schema') with sink ('node-urls'='%s')",
+                receiverEnv.getDataNodeWrapperList().get(0).getIpAndPortString())),
+        BaseEnv.TABLE_SQL_DIALECT);
+
+    TestUtils.assertDataAlwaysOnEnv(
+        receiverEnv,
+        "show devices from table1",
+        "a,b,c,",
+        new HashSet<>(Arrays.asList("1,1,1,", "2,2,2,")),
+        "test");
+  }
 }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/schema/CreateOrUpdateTableDeviceNode.java

@@ -31,6 +31,7 @@
 import org.apache.iotdb.db.schemaengine.schemaregion.SchemaRegionPlanVisitor;
 
 import org.apache.tsfile.file.metadata.IDeviceID;
+import org.apache.tsfile.utils.Constants;
 import org.apache.tsfile.utils.ReadWriteIOUtils;
 
 import java.io.DataOutputStream;
@@ -215,7 +216,7 @@ protected void serializeAttributes(final DataOutputStream stream) throws IOExcep
         ReadWriteIOUtils.writeObject(value, stream);
       }
       for (int i = 0; i < attributeNameList.size() - deviceAttributeValueList.length; ++i) {
-        ReadWriteIOUtils.writeObject(null, stream);
+        ReadWriteIOUtils.writeObject(Constants.NONE, stream);
       }
     }
   }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/attribute/DeviceAttributeStore.java

@@ -27,6 +27,7 @@
 import org.apache.iotdb.db.schemaengine.schemaregion.attribute.update.UpdateDetailContainer;
 
 import org.apache.tsfile.utils.Binary;
+import org.apache.tsfile.utils.Constants;
 import org.apache.tsfile.utils.RamUsageEstimator;
 import org.apache.tsfile.utils.ReadWriteIOUtils;
 import org.slf4j.Logger;
@@ -163,6 +164,12 @@ public Map<String, Binary> alterAttribute(
     final Map<String, Binary> attributeMap = deviceAttributeList.get(pointer);
     for (int i = 0; i < nameList.size(); i++) {
       final String key = nameList.get(i);
+      if (valueList.length <= i) {
+        break;
+      }
+      if (valueList[i] == Constants.NONE) {
+        continue;
+      }
       final Binary value = (Binary) valueList[i];
 
       originMemUsage =

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/tools/schema/SRStatementGenerator.java

@@ -45,6 +45,7 @@
 import org.apache.iotdb.db.schemaengine.schemaregion.mtree.impl.mem.snapshot.MemMTreeSnapshotUtil;
 
 import org.apache.tsfile.utils.Binary;
+import org.apache.tsfile.utils.Constants;
 import org.apache.tsfile.utils.Pair;
 import org.apache.tsfile.utils.ReadWriteIOUtils;
 import org.slf4j.Logger;
@@ -447,7 +448,13 @@ private List<Object> genActivateTemplateOrUpdateDeviceStatement(
           attributeNameList = new ArrayList<>(tableAttributes.keySet());
         }
         final List<Object> attributeValues =
-            attributeNameList.stream().map(tableAttributes::remove).collect(Collectors.toList());
+            attributeNameList.stream()
+                .map(
+                    attributeKey -> {
+                      final Object attributeValue = tableAttributes.remove(attributeKey);
+                      return Objects.nonNull(attributeValue) ? attributeValue : Constants.NONE;
+                    })
+                .collect(Collectors.toList());
         tableAttributes.forEach(
             (attributeKey, attributeValue) -> {
               attributeNameList.add(attributeKey);

iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/SchemaRegionSnapshotParserTest.java

@@ -48,6 +48,7 @@
 import org.apache.tsfile.file.metadata.enums.CompressionType;
 import org.apache.tsfile.file.metadata.enums.TSEncoding;
 import org.apache.tsfile.utils.Binary;
+import org.apache.tsfile.utils.Constants;
 import org.apache.tsfile.utils.Pair;
 import org.junit.After;
 import org.junit.Assert;
@@ -313,7 +314,9 @@ public void testTableDeviceAttributeTranslateSnapshot() throws Exception {
                     Arrays.asList("a", "c"),
                     Arrays.asList(
                         new Binary[] {new Binary("b", TSFileConfig.STRING_CHARSET)},
-                        new Binary[] {null, new Binary("d", TSFileConfig.STRING_CHARSET)})),
+                        new Object[] {
+                          Constants.NONE, new Binary("d", TSFileConfig.STRING_CHARSET)
+                        })),
                 new CreateOrUpdateDevice(
                     databasePath.getNodes()[1],
                     anotherTable,

pom.xml

@@ -163,7 +163,7 @@
         <thrift.version>0.14.1</thrift.version>
         <xz.version>1.9</xz.version>
         <zstd-jni.version>1.5.6-3</zstd-jni.version>
-        <tsfile.version>2.2.1-260327-SNAPSHOT</tsfile.version>
+        <tsfile.version>2.3.0-260422-SNAPSHOT</tsfile.version>
     </properties>
     <!--
     if we claim dependencies in dependencyManagement, then we do not claim