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

* Fix partial insert handling for null measurements (#17879)

* Fix partial insert handling for null measurements

* Fix partial insert handling for missing columns

* Fix table insert schema validation for partial insert

* Fix partial insert null column handling

* Fix failed measurement WAL serialization tests

* Fix all-failed tablet test column reuse

(cherry picked from commit feaeaa6)

* spotless

* Fix ITs

* Handle null values in type inference

* Share WAL measurement schema serialization

Author: Caideyipi

integration-test/src/test/java/org/apache/iotdb/session/it/IoTDBSessionSimpleIT.java

@@ -1185,13 +1185,14 @@ public void insertRecordsWithDuplicatedMeasurementsTest() {
   public void insertRecordsWithExpiredDataTest()
       throws IoTDBConnectionException, StatementExecutionException {
     try (ISession session = EnvFactory.getEnv().getSessionConnection()) {
+      String device = "root.sg.expiredData.d1";
       List<Long> times = new ArrayList<>();
       List<List<String>> measurements = new ArrayList<>();
       List<List<TSDataType>> datatypes = new ArrayList<>();
       List<List<Object>> values = new ArrayList<>();
       List<String> devices = new ArrayList<>();
 
-      devices.add("root.sg.d1");
+      devices.add(device);
       addLine(
           times,
           measurements,
@@ -1204,14 +1205,14 @@ public void insertRecordsWithExpiredDataTest()
           TSDataType.INT32,
           1,
           2);
-      session.executeNonQueryStatement("set ttl to root.sg.d1 1");
-      try {
-        session.insertRecords(devices, times, measurements, datatypes, values);
-      } catch (Exception e) {
-        assertTrue(e.getMessage().contains("less than ttl time bound"));
-      }
-      session.executeNonQueryStatement("unset ttl to root.sg.d1");
-      SessionDataSet dataSet = session.executeQueryStatement("select * from root.sg.d1");
+      session.executeNonQueryStatement("set ttl to " + device + " 1");
+      StatementExecutionException exception =
+          assertThrows(
+              StatementExecutionException.class,
+              () -> session.insertRecords(devices, times, measurements, datatypes, values));
+      assertTrue(exception.getMessage().contains("less than ttl time bound"));
+      session.executeNonQueryStatement("unset ttl to " + device);
+      SessionDataSet dataSet = session.executeQueryStatement("select * from " + device);
       assertFalse(dataSet.hasNext());
     }
   }

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

@@ -385,8 +385,11 @@ private void generateColumnIndexMapper(
     // 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++;
+        }
       }
     }
 

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

@@ -100,7 +100,13 @@ protected boolean checkAndCastDataType(int columnIndex, TSDataType dataType) {
 
   @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 @@ public void transferType(ZoneId zoneId) throws QueryProcessException {
 
       // 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

@@ -510,6 +510,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)
@@ -521,7 +524,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()) {
@@ -585,6 +591,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)
@@ -596,7 +605,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/schema/NormalSchemaFetcher.java

@@ -337,7 +337,9 @@ void processNormalTimeSeries(
           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,24 +318,20 @@ 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++) {
-      // ignore failed partial insert
-      if (measurements[i] == null) {
-        continue;
+    for (int i = 0; measurements != null && i < measurements.length; i++) {
+      if (shouldSerializeMeasurementToWAL(i)) {
+        byteLen += WALWriteUtils.sizeToWrite(measurementSchemas[i]);
       }
-      byteLen += WALWriteUtils.sizeToWrite(measurementSchemas[i]);
     }
     return byteLen;
   }
 
   /** Serialize measurement schemas, ignoring failed time series */
   protected void serializeMeasurementSchemasToWAL(IWALByteBufferView buffer) {
-    for (int i = 0; i < measurements.length; i++) {
-      // ignore failed partial insert
-      if (measurements[i] == null) {
-        continue;
+    for (int i = 0; measurements != null && i < measurements.length; i++) {
+      if (shouldSerializeMeasurementToWAL(i)) {
+        WALWriteUtils.write(measurementSchemas[i], buffer);
       }
-      WALWriteUtils.write(measurementSchemas[i], buffer);
     }
   }
 
@@ -373,8 +369,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,11 +385,46 @@ 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 (shouldSerializeMeasurement(i)) {
+        validMeasurementNumber++;
+      }
     }
-    return true;
+    return validMeasurementNumber;
+  }
+
+  protected int getValidMeasurementNumberForWAL() {
+    int validMeasurementNumber = 0;
+    for (int i = 0; measurements != null && i < measurements.length; i++) {
+      if (shouldSerializeMeasurementToWAL(i)) {
+        validMeasurementNumber++;
+      }
+    }
+    return validMeasurementNumber;
+  }
+
+  protected boolean shouldSerializeMeasurement(final int index) {
+    return !isMeasurementFailed(index);
+  }
+
+  protected boolean shouldSerializeMeasurementToWAL(final int index) {
+    return shouldSerializeMeasurement(index)
+        && measurementSchemas != null
+        && index < measurementSchemas.length
+        && measurementSchemas[index] != null;
+  }
+
+  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() {