[To dev/1.3] Fix partial insert handling for null measurements (#17879)

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

@@ -371,7 +371,7 @@
     }
   }
 
   private void generateColumnIndexMapper(
       final String[] originMeasurementList,
       final PipePattern pattern,
       final Integer[] originColumnIndex2FilteredColumnIndexMapperList) {
@@ -380,8 +380,11 @@
     // case 1: for example, pattern is root.a.b or pattern is null and device is root.a.b.c
     // in this case, all data can be matched without checking the measurements
     if (Objects.isNull(pattern) || pattern.isRoot() || pattern.coversDevice(deviceId)) {
+      int filteredCount = 0;
       for (int i = 0; i < originColumnSize; i++) {
-        originColumnIndex2FilteredColumnIndexMapperList[i] = i;
+        if (originMeasurementList[i] != null) {
+          originColumnIndex2FilteredColumnIndexMapperList[i] = filteredCount++;
+        }
       }
     }
 
@@ -435,7 +438,7 @@
     return IntStream.range(0, rowCount).boxed().collect(Collectors.toList());
   }
 
   private static Object filterValueColumnsByRowIndexList(
       @NonNull final TSDataType type,
       @NonNull final Object originValueColumn,
       @NonNull final List<Integer> rowIndexList,

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

@@ -99,8 +99,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()) {
@@ -119,6 +125,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

@@ -488,6 +488,9 @@ private static long sizeOfBitMap(final BitMap bitMaps) {
 
   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)
@@ -499,7 +502,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()) {
@@ -563,6 +569,9 @@ private static long getBinarySize(final Binary[] binaries) {
 
   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)
@@ -574,7 +583,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/queryengine/plan/analyze/cache/schema/DataNodeSchemaCache.java

@@ -166,17 +166,24 @@ public ClusterSchemaTree getMatchedNormalSchema(final PartialPath fullPath) {
   public List<Integer> computeWithoutTemplate(final ISchemaComputation schemaComputation) {
     final List<Integer> indexOfMissingMeasurements = new ArrayList<>();
     final String[] measurements = schemaComputation.getMeasurements();
+    if (measurements == null) {
+      return indexOfMissingMeasurements;
+    }
 
     final IDeviceSchema schema =
         deviceSchemaCache.getDeviceSchema(schemaComputation.getDevicePath());
     if (!(schema instanceof DeviceNormalSchema)) {
-      return IntStream.range(0, schemaComputation.getMeasurements().length)
+      return IntStream.range(0, measurements.length)
+          .filter(i -> measurements[i] != null)
           .boxed()
           .collect(Collectors.toList());
     }
     final DeviceNormalSchema treeSchema = (DeviceNormalSchema) schema;
 
-    for (int i = 0; i < schemaComputation.getMeasurements().length; i++) {
+    for (int i = 0; i < measurements.length; i++) {
+      if (measurements[i] == null) {
+        continue;
+      }
       final SchemaCacheEntry value = treeSchema.getSchemaCacheEntry(measurements[i]);
       if (value == null) {
         indexOfMissingMeasurements.add(i);

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

@@ -67,7 +67,9 @@ public int update(final String[] measurements, final IMeasurementSchema[] schema
 
     for (int i = 0; i < length; ++i) {
       // Skip this to avoid instance creation/gc for writing performance
-      if (measurements[i] == null || measurementMap.containsKey(measurements[i])) {
+      if (measurements[i] == null
+          || schemas[i] == null
+          || measurementMap.containsKey(measurements[i])) {
         continue;
       }
       diff += putEntry(measurements[i], schemas[i], null);

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

@@ -188,7 +188,7 @@
     return indexOfMissingMeasurements;
   }
 
   void processNormalTimeSeries(
       List<? extends ISchemaComputationWithAutoCreation> schemaComputationWithAutoCreationList,
       MPPQueryContext context) {
     // [Step 0] Record the input value.
@@ -337,7 +337,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);
                     }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertNode.java

@@ -168,7 +168,7 @@ public TSDataType[] getDataTypes() {
   }
 
   public TSDataType getDataType(int index) {
-    return dataTypes[index];
+    return dataTypes == null || index < 0 || index >= dataTypes.length ? null : dataTypes[index];
   }
 
   public void setDataTypes(TSDataType[] dataTypes) {
@@ -318,9 +318,12 @@ private static Map<String, String> deserializeProps(ByteBuffer buffer) {
   /** Serialized size of measurement schemas, ignoring failed time series */
   protected int serializeMeasurementSchemasSize() {
     int byteLen = 0;
-    for (int i = 0; i < measurements.length; i++) {
+    for (int i = 0; measurements != null && i < measurements.length; i++) {
       // ignore failed partial insert
-      if (measurements[i] == null) {
+      if (measurements[i] == null
+          || measurementSchemas == null
+          || i >= measurementSchemas.length
+          || measurementSchemas[i] == null) {
         continue;
       }
       byteLen += WALWriteUtils.sizeToWrite(measurementSchemas[i]);
@@ -330,9 +333,12 @@ protected int serializeMeasurementSchemasSize() {
 
   /** Serialize measurement schemas, ignoring failed time series */
   protected void serializeMeasurementSchemasToWAL(IWALByteBufferView buffer) {
-    for (int i = 0; i < measurements.length; i++) {
+    for (int i = 0; measurements != null && i < measurements.length; i++) {
       // ignore failed partial insert
-      if (measurements[i] == null) {
+      if (measurements[i] == null
+          || measurementSchemas == null
+          || i >= measurementSchemas.length
+          || measurementSchemas[i] == null) {
         continue;
       }
       WALWriteUtils.write(measurementSchemas[i], buffer);
@@ -373,8 +379,8 @@ public void markFailedMeasurement(int index) {
   }
 
   public boolean hasValidMeasurements() {
-    for (Object o : measurements) {
-      if (o != null) {
+    for (int i = 0; measurements != null && i < measurements.length; i++) {
+      if (!isMeasurementFailed(i)) {
         return true;
       }
     }
@@ -389,19 +395,35 @@ public int getFailedMeasurementNumber() {
     return failedMeasurementNumber;
   }
 
-  public boolean allMeasurementFailed() {
-    if (measurements != null) {
-      return failedMeasurementNumber >= measurements.length;
+  protected int getValidMeasurementNumber() {
+    int validMeasurementNumber = 0;
+    for (int i = 0; measurements != null && i < measurements.length; i++) {
+      if (!isMeasurementFailed(i)) {
+        validMeasurementNumber++;
+      }
     }
-    return true;
+    return validMeasurementNumber;
+  }
+
+  public boolean isMeasurementFailed(int index) {
+    return measurements == null
+        || index < 0
+        || index >= measurements.length
+        || measurements[index] == null;
+  }
+
+  public boolean allMeasurementFailed() {
+    return measurements == null || !hasValidMeasurements();
   }
 
   public String[] getRawMeasurements() {
     String[] measurements = getMeasurements();
     MeasurementSchema[] measurementSchemas = getMeasurementSchemas();
     String[] rawMeasurements = new String[measurements.length];
     for (int i = 0; i < measurements.length; i++) {
-      if (measurementSchemas[i] != null) {
+      if (measurementSchemas != null
+          && i < measurementSchemas.length
+          && measurementSchemas[i] != null) {
         // get raw measurement rather than alias
         rawMeasurements[i] = measurementSchemas[i].getMeasurementId();
       } else {