Fix partial insert handling for null measurements

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

@@ -324,7 +324,7 @@ private void prepareTypeConversionTest(
         generateMeasurementSchemas();
 
     // Generate createTimeSeries in sender and receiver
-    String uuid = "bcdedit";
+    String uuid = "bcdedit" + Long.toHexString(System.nanoTime());
     for (Pair<MeasurementSchema, MeasurementSchema> pair : measurementSchemas) {
       createTimeSeries(uuid, pair.left.getMeasurementName(), pair.left.getType().name(), senderEnv);
       createTimeSeries(

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/parser/TabletInsertionEventParser.java

@@ -54,6 +54,7 @@
 import java.util.Map;
 import java.util.Objects;
 import java.util.function.BiConsumer;
+import java.util.function.IntPredicate;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 
@@ -114,8 +115,10 @@
 
   //////////////////////////// parse ////////////////////////////
 
   protected void parse(final InsertRowNode insertRowNode) {
-    final int originColumnSize = insertRowNode.getMeasurements().length;
+    final String[] originColumnNameStringList = insertRowNode.getMeasurements();
+    final int originColumnSize =
+        originColumnNameStringList == null ? 0 : originColumnNameStringList.length;
     final Integer[] originColumnIndex2FilteredColumnIndexMapperList = new Integer[originColumnSize];
 
     // The full path is always cached when device path is deserialized
@@ -127,14 +130,25 @@
     final List<Integer> rowIndexList = generateRowIndexList(originTimestampColumn);
     this.timestampColumn = rowIndexList.stream().mapToLong(i -> originTimestampColumn[i]).toArray();
 
+    final MeasurementSchema[] originMeasurementSchemaList = insertRowNode.getMeasurementSchemas();
+    final TsTableColumnCategory[] originColumnCategories = insertRowNode.getColumnCategories();
+    final TSDataType[] originValueDataTypes = insertRowNode.getDataTypes();
+    final Object[] originValues = insertRowNode.getValues();
+
     generateColumnIndexMapper(
-        insertRowNode.getMeasurements(), originColumnIndex2FilteredColumnIndexMapperList);
+        originColumnNameStringList, originColumnIndex2FilteredColumnIndexMapperList);
 
     final int filteredColumnSize =
-        Arrays.stream(originColumnIndex2FilteredColumnIndexMapperList)
-            .filter(Objects::nonNull)
-            .toArray()
-            .length;
+        compactColumnIndexMapper(
+            originColumnIndex2FilteredColumnIndexMapperList,
+            i ->
+                isValidOriginColumn(
+                        originColumnNameStringList,
+                        originMeasurementSchemaList,
+                        originValueDataTypes,
+                        i)
+                    && originValues != null
+                    && i < originValues.length);
 
     this.measurementSchemaList = new MeasurementSchema[filteredColumnSize];
     this.columnNameStringList = new String[filteredColumnSize];
@@ -143,19 +157,15 @@
     this.valueColumns = new Object[filteredColumnSize];
     this.nullValueColumnBitmaps = new BitMap[filteredColumnSize];
 
-    final MeasurementSchema[] originMeasurementSchemaList = insertRowNode.getMeasurementSchemas();
-    final String[] originColumnNameStringList = insertRowNode.getMeasurements();
-    final TsTableColumnCategory[] originColumnCategories = insertRowNode.getColumnCategories();
-    final TSDataType[] originValueDataTypes = insertRowNode.getDataTypes();
-    final Object[] originValues = insertRowNode.getValues();
-
     for (int i = 0; i < originColumnIndex2FilteredColumnIndexMapperList.length; i++) {
       if (originColumnIndex2FilteredColumnIndexMapperList[i] != null) {
         final int filteredColumnIndex = originColumnIndex2FilteredColumnIndexMapperList[i];
         this.measurementSchemaList[filteredColumnIndex] = originMeasurementSchemaList[i];
         this.columnNameStringList[filteredColumnIndex] = originColumnNameStringList[i];
         this.valueColumnTypes[filteredColumnIndex] =
-            originColumnCategories != null && originColumnCategories[i] != null
+            originColumnCategories != null
+                    && i < originColumnCategories.length
+                    && originColumnCategories[i] != null
                 ? originColumnCategories[i].toTsFileColumnType()
                 : ColumnCategory.FIELD;
         this.valueColumnDataTypes[filteredColumnIndex] = originValueDataTypes[i];
@@ -189,8 +199,10 @@
     }
   }
 
   protected void parse(final InsertTabletNode insertTabletNode) throws IllegalPathException {
-    final int originColumnSize = insertTabletNode.getMeasurements().length;
+    final String[] originColumnNameStringList = insertTabletNode.getMeasurements();
+    final int originColumnSize =
+        originColumnNameStringList == null ? 0 : originColumnNameStringList.length;
     final Integer[] originColumnIndex2FilteredColumnIndexMapperList = new Integer[originColumnSize];
 
     // The full path is always cached when device path is deserialized
@@ -202,14 +214,28 @@
     final List<Integer> rowIndexList = generateRowIndexList(originTimestampColumn);
     this.timestampColumn = rowIndexList.stream().mapToLong(i -> originTimestampColumn[i]).toArray();
 
+    final MeasurementSchema[] originMeasurementSchemaList =
+        insertTabletNode.getMeasurementSchemas();
+    final TsTableColumnCategory[] originColumnCategories = insertTabletNode.getColumnCategories();
+    final TSDataType[] originValueColumnDataTypes = insertTabletNode.getDataTypes();
+    final Object[] originValueColumns = insertTabletNode.getColumns();
+    final BitMap[] originBitMapList = insertTabletNode.getBitMaps();
+
     generateColumnIndexMapper(
-        insertTabletNode.getMeasurements(), originColumnIndex2FilteredColumnIndexMapperList);
+        originColumnNameStringList, originColumnIndex2FilteredColumnIndexMapperList);
 
     final int filteredColumnSize =
-        Arrays.stream(originColumnIndex2FilteredColumnIndexMapperList)
-            .filter(Objects::nonNull)
-            .toArray()
-            .length;
+        compactColumnIndexMapper(
+            originColumnIndex2FilteredColumnIndexMapperList,
+            i ->
+                isValidOriginColumn(
+                        originColumnNameStringList,
+                        originMeasurementSchemaList,
+                        originValueColumnDataTypes,
+                        i)
+                    && originValueColumns != null
+                    && i < originValueColumns.length
+                    && originValueColumns[i] != null);
 
     this.measurementSchemaList = new MeasurementSchema[filteredColumnSize];
     this.columnNameStringList = new String[filteredColumnSize];
@@ -218,21 +244,15 @@
     this.valueColumns = new Object[filteredColumnSize];
     this.nullValueColumnBitmaps = new BitMap[filteredColumnSize];
 
-    final MeasurementSchema[] originMeasurementSchemaList =
-        insertTabletNode.getMeasurementSchemas();
-    final String[] originColumnNameStringList = insertTabletNode.getMeasurements();
-    final TsTableColumnCategory[] originColumnCategories = insertTabletNode.getColumnCategories();
-    final TSDataType[] originValueColumnDataTypes = insertTabletNode.getDataTypes();
-    final Object[] originValueColumns = insertTabletNode.getColumns();
-    final BitMap[] originBitMapList = insertTabletNode.getBitMaps();
-
     for (int i = 0; i < originColumnIndex2FilteredColumnIndexMapperList.length; i++) {
       if (originColumnIndex2FilteredColumnIndexMapperList[i] != null) {
         final int filteredColumnIndex = originColumnIndex2FilteredColumnIndexMapperList[i];
         this.measurementSchemaList[filteredColumnIndex] = originMeasurementSchemaList[i];
         this.columnNameStringList[filteredColumnIndex] = originColumnNameStringList[i];
         this.valueColumnTypes[filteredColumnIndex] =
-            originColumnCategories != null && originColumnCategories[i] != null
+            originColumnCategories != null
+                    && i < originColumnCategories.length
+                    && originColumnCategories[i] != null
                 ? originColumnCategories[i].toTsFileColumnType()
                 : ColumnCategory.FIELD;
         this.valueColumnDataTypes[filteredColumnIndex] = originValueColumnDataTypes[i];
@@ -298,10 +318,9 @@
         originMeasurementList, originColumnIndex2FilteredColumnIndexMapperList);
 
     final int filteredColumnSize =
-        Arrays.stream(originColumnIndex2FilteredColumnIndexMapperList)
-            .filter(Objects::nonNull)
-            .toArray()
-            .length;
+        compactColumnIndexMapper(
+            originColumnIndex2FilteredColumnIndexMapperList,
+            i -> isValidOriginColumn(originMeasurementSchemaList, i));
 
     this.measurementSchemaList = new MeasurementSchema[filteredColumnSize];
     this.columnNameStringList = new String[filteredColumnSize];
@@ -373,6 +392,46 @@
       final String[] originMeasurementList,
       final Integer[] originColumnIndex2FilteredColumnIndexMapperList);
 
+  private static int compactColumnIndexMapper(
+      final Integer[] originColumnIndex2FilteredColumnIndexMapperList,
+      final IntPredicate columnValidator) {
+    int filteredCount = 0;
+    for (int i = 0; i < originColumnIndex2FilteredColumnIndexMapperList.length; i++) {
+      if (originColumnIndex2FilteredColumnIndexMapperList[i] != null && columnValidator.test(i)) {
+        originColumnIndex2FilteredColumnIndexMapperList[i] = filteredCount++;
+      } else {
+        originColumnIndex2FilteredColumnIndexMapperList[i] = null;
+      }
+    }
+    return filteredCount;
+  }
+
+  private static boolean isValidOriginColumn(
+      final String[] originColumnNameStringList,
+      final MeasurementSchema[] originMeasurementSchemaList,
+      final TSDataType[] originValueDataTypes,
+      final int index) {
+    return originColumnNameStringList != null
+        && index < originColumnNameStringList.length
+        && originColumnNameStringList[index] != null
+        && originMeasurementSchemaList != null
+        && index < originMeasurementSchemaList.length
+        && originMeasurementSchemaList[index] != null
+        && originMeasurementSchemaList[index].getType() != null
+        && originValueDataTypes != null
+        && index < originValueDataTypes.length
+        && originValueDataTypes[index] != null;
+  }
+
+  private static boolean isValidOriginColumn(
+      final List<IMeasurementSchema> originMeasurementSchemaList, final int index) {
+    return originMeasurementSchemaList != null
+        && index < originMeasurementSchemaList.size()
+        && originMeasurementSchemaList.get(index) != null
+        && originMeasurementSchemaList.get(index).getMeasurementName() != null
+        && originMeasurementSchemaList.get(index).getType() != null;
+  }
+
   private List<Integer> generateRowIndexList(final long[] originTimestampColumn) {
     final int rowCount = originTimestampColumn.length;
     if (Objects.isNull(sourceEvent) || !sourceEvent.shouldParseTime()) {
@@ -403,7 +462,7 @@
     return IntStream.range(0, rowCount).boxed().collect(Collectors.toList());
   }
 
   private static Object filterValueColumnsByRowIndexList(
       final TSDataType type,
       final Object originValueColumn,
       final List<Integer> rowIndexList,

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/transform/statement/PipeConvertedInsertRowStatement.java

@@ -106,8 +106,14 @@
   }
 
   @Override
   public void transferType(ZoneId zoneId) throws QueryProcessException {
+    if (measurementSchemas == null) {
+      return;
+    }
     for (int i = 0; i < measurementSchemas.length; i++) {
+      if (!isColumnPresent(i) || dataTypes == null || i >= dataTypes.length) {
+        continue;
+      }
       // null when time series doesn't exist
       if (measurementSchemas[i] == null) {
         if (!IoTDBDescriptor.getInstance().getConfig().isEnablePartialInsert()) {
@@ -126,6 +132,9 @@
 
       // parse string value to specific type
       dataTypes[i] = measurementSchemas[i].getType();
+      if (values == null || i >= values.length || values[i] == null) {
+        continue;
+      }
       try {
         values[i] = ValueConverter.parse(values[i].toString(), dataTypes[i]);
       } catch (Exception e) {

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/InsertNodeMemoryEstimator.java

@@ -548,6 +548,9 @@ private static long sizeOfBinary(final Binary binary) {
 
   public static long sizeOfColumns(
       final Object[] columns, final MeasurementSchema[] measurementSchemas) {
+    if (Objects.isNull(columns)) {
+      return 0L;
+    }
     // Directly calculate if measurementSchemas are absent
     if (Objects.isNull(measurementSchemas)) {
       return RamUsageEstimator.shallowSizeOf(columns)
@@ -559,7 +562,10 @@ public static long sizeOfColumns(
         RamUsageEstimator.alignObjectSize(
             NUM_BYTES_ARRAY_HEADER + NUM_BYTES_OBJECT_REF * columns.length);
     for (int i = 0; i < columns.length; i++) {
-      if (measurementSchemas[i] == null || measurementSchemas[i].getType() == null) {
+      if (columns[i] == null
+          || i >= measurementSchemas.length
+          || measurementSchemas[i] == null
+          || measurementSchemas[i].getType() == null) {
         continue;
       }
       switch (measurementSchemas[i].getType()) {
@@ -611,6 +617,9 @@ private static long getNumBytesUnknownObject(final Object obj) {
 
   public static long sizeOfValues(
       final Object[] values, final MeasurementSchema[] measurementSchemas) {
+    if (Objects.isNull(values)) {
+      return 0L;
+    }
     // Directly calculate if measurementSchemas are absent
     if (Objects.isNull(measurementSchemas)) {
       return RamUsageEstimator.shallowSizeOf(values)
@@ -622,7 +631,9 @@ public static long sizeOfValues(
         RamUsageEstimator.alignObjectSize(
             NUM_BYTES_ARRAY_HEADER + NUM_BYTES_OBJECT_REF * values.length);
     for (int i = 0; i < values.length; i++) {
-      if (measurementSchemas[i] == null || measurementSchemas[i].getType() == null) {
+      if (i >= measurementSchemas.length
+          || measurementSchemas[i] == null
+          || measurementSchemas[i].getType() == null) {
         size += NUM_BYTES_OBJECT_HEADER;
         continue;
       }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/TabletStatementConverter.java

@@ -383,7 +383,7 @@
    * @return Pair of values array and memory size
    */
   @SuppressWarnings("squid:S3776") // Suppress high Cognitive Complexity warning
   private static Pair<Object[], Long> readValuesFromBufferWithMemory(
       final ByteBuffer byteBuffer, final TSDataType[] types, final int columns, final int rowSize) {
     final Object[] values = new Object[columns];
 
@@ -395,7 +395,7 @@
     for (int i = 0; i < columns; i++) {
       final boolean isValueColumnsNotNull =
           BytesUtils.byteToBool(ReadWriteIOUtils.readByte(byteBuffer));
-      if (isValueColumnsNotNull && types[i] == null) {
+      if (types[i] == null) {
         continue;
       }
 

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/schema/NormalSchemaFetcher.java

@@ -196,7 +196,7 @@
     return indexOfMissingMeasurements;
   }
 
   void processNormalTimeSeries(
       List<? extends ISchemaComputationWithAutoCreation> schemaComputationWithAutoCreationList,
       MPPQueryContext context) {
     // [Step 0] Record the input value.
@@ -358,7 +358,9 @@
           schemaComputationWithAutoCreationList.stream()
               .map(
                   o -> {
-                    TSDataType[] dataTypes = new TSDataType[o.getMeasurements().length];
+                    final String[] measurements = o.getMeasurements();
+                    TSDataType[] dataTypes =
+                        new TSDataType[measurements == null ? 0 : measurements.length];
                     for (int i = 0, length = dataTypes.length; i < length; i++) {
                       dataTypes[i] = o.getDataType(i);
                     }