[To dev/1.3] Load: Support auto data type conversion when data type mismatch detected during analysis stage (#14529)

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

@@ -31,6 +31,7 @@
 import org.apache.tsfile.enums.TSDataType;
 import org.apache.tsfile.file.metadata.enums.TSEncoding;
 import org.apache.tsfile.read.common.Path;
+import org.apache.tsfile.utils.Pair;
 import org.apache.tsfile.write.schema.MeasurementSchema;
 import org.junit.After;
 import org.junit.Assert;
@@ -47,6 +48,7 @@
 import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.sql.Statement;
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
@@ -55,6 +57,7 @@
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
 
 import static org.apache.iotdb.db.it.utils.TestUtils.assertNonQueryTestFail;
 import static org.apache.iotdb.db.it.utils.TestUtils.createUser;
@@ -897,6 +900,74 @@ public void testLoadLocally() throws Exception {
     }
   }
 
+  @Test
+  public void testLoadWithConvertOnTypeMismatch() throws Exception {
+
+    List<Pair<MeasurementSchema, MeasurementSchema>> measurementSchemas =
+        generateMeasurementSchemasForDataTypeConvertion();
+
+    final File file = new File(tmpDir, "1-0-0-0.tsfile");
+
+    long writtenPoint = 0;
+    List<MeasurementSchema> schemaList1 =
+        measurementSchemas.stream().map(pair -> pair.left).collect(Collectors.toList());
+    List<MeasurementSchema> schemaList2 =
+        measurementSchemas.stream().map(pair -> pair.right).collect(Collectors.toList());
+
+    try (final TsFileGenerator generator = new TsFileGenerator(file)) {
+      generator.registerTimeseries(SchemaConfig.DEVICE_0, schemaList2);
+
+      generator.generateData(SchemaConfig.DEVICE_0, 100, PARTITION_INTERVAL / 10_000, false);
+
+      writtenPoint = generator.getTotalNumber();
+    }
+
+    try (final Connection connection = EnvFactory.getEnv().getConnection();
+        final Statement statement = connection.createStatement()) {
+
+      for (MeasurementSchema schema : schemaList1) {
+        statement.execute(convert2SQL(SchemaConfig.DEVICE_0, schema));
+      }
+
+      statement.execute(String.format("load \"%s\" ", file.getAbsolutePath()));
+
+      try (final ResultSet resultSet =
+          statement.executeQuery("select count(*) from root.** group by level=1,2")) {
+        if (resultSet.next()) {
+          final long sgCount = resultSet.getLong("count(root.sg.test_0.*.*)");
+          Assert.assertEquals(writtenPoint, sgCount);
+        } else {
+          Assert.fail("This ResultSet is empty.");
+        }
+      }
+    }
+  }
+
+  private List<Pair<MeasurementSchema, MeasurementSchema>>
+      generateMeasurementSchemasForDataTypeConvertion() {
+    TSDataType[] dataTypes = {
+      TSDataType.STRING,
+      TSDataType.TEXT,
+      TSDataType.BLOB,
+      TSDataType.TIMESTAMP,
+      TSDataType.BOOLEAN,
+      TSDataType.DATE,
+      TSDataType.DOUBLE,
+      TSDataType.FLOAT,
+      TSDataType.INT32,
+      TSDataType.INT64
+    };
+    List<Pair<MeasurementSchema, MeasurementSchema>> pairs = new ArrayList<>();
+
+    for (TSDataType type : dataTypes) {
+      for (TSDataType dataType : dataTypes) {
+        String id = String.format("%s2%s", type.name(), dataType.name());
+        pairs.add(new Pair<>(new MeasurementSchema(id, type), new MeasurementSchema(id, dataType)));
+      }
+    }
+    return pairs;
+  }
+
   private static class SchemaConfig {
     private static final String STORAGE_GROUP_0 = "root.sg.test_0";
     private static final String STORAGE_GROUP_1 = "root.sg.test_1";

iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TSStatusCode.java

@@ -173,6 +173,9 @@ public enum TSStatusCode {
   PIPE_ERROR(1107),
   PIPESERVER_ERROR(1108),
   VERIFY_METADATA_ERROR(1109),
+  LOAD_TEMPORARY_UNAVAILABLE_EXCEPTION(1110),
+  LOAD_IDEMPOTENT_CONFLICT_EXCEPTION(1111),
+  LOAD_USER_CONFLICT_EXCEPTION(1112),
 
   // UDF
   UDF_LOAD_CLASS_ERROR(1200),

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/exception/VerifyMetadataException.java

@@ -23,16 +23,12 @@
 import org.apache.iotdb.rpc.TSStatusCode;
 
 public class VerifyMetadataException extends IoTDBException {
-  public VerifyMetadataException(
-      String path, String compareInfo, String tsFileInfo, String tsFilePath, String IoTDBInfo) {
-    super(
-        String.format(
-            "%s %s mismatch, %s in tsfile %s, but %s in IoTDB.",
-            path, compareInfo, tsFileInfo, tsFilePath, IoTDBInfo),
-        TSStatusCode.VERIFY_METADATA_ERROR.getStatusCode());
-  }
 
   public VerifyMetadataException(String message) {
     super(message, TSStatusCode.VERIFY_METADATA_ERROR.getStatusCode());
   }
+
+  public VerifyMetadataException(String message, int errorCode) {
+    super(message, errorCode);
+  }
 }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/exception/VerifyMetadataTypeMismatchException.java

@@ -0,0 +1,29 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.exception;
+
+import org.apache.iotdb.rpc.TSStatusCode;
+
+public class VerifyMetadataTypeMismatchException extends VerifyMetadataException {
+
+  public VerifyMetadataTypeMismatchException(String message) {
+    super(message, TSStatusCode.VERIFY_METADATA_ERROR.getStatusCode());
+  }
+}

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/legacy/loader/TsFileLoader.java

@@ -56,6 +56,7 @@ public void load() {
     try {
       LoadTsFileStatement statement = new LoadTsFileStatement(tsFile.getAbsolutePath());
       statement.setDeleteAfterLoad(true);
+      statement.setConvertOnTypeMismatch(true);
       statement.setDatabaseLevel(parseSgLevel());
       statement.setVerifySchema(true);
       statement.setAutoCreateDatabase(false);

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java

@@ -479,6 +479,7 @@ private TSStatus loadTsFileSync(final String fileAbsolutePath) throws FileNotFou
     final LoadTsFileStatement statement = new LoadTsFileStatement(fileAbsolutePath);
 
     statement.setDeleteAfterLoad(true);
+    statement.setConvertOnTypeMismatch(true);
     statement.setVerifySchema(true);
     statement.setAutoCreateDatabase(false);
 

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

@@ -36,6 +36,7 @@
 import org.apache.iotdb.db.conf.IoTDBConfig;
 import org.apache.iotdb.db.conf.IoTDBDescriptor;
 import org.apache.iotdb.db.exception.VerifyMetadataException;
+import org.apache.iotdb.db.exception.VerifyMetadataTypeMismatchException;
 import org.apache.iotdb.db.exception.load.LoadEmptyFileException;
 import org.apache.iotdb.db.exception.load.LoadFileException;
 import org.apache.iotdb.db.exception.load.LoadReadOnlyException;
@@ -63,6 +64,7 @@
 import org.apache.iotdb.db.storageengine.dataregion.tsfile.timeindex.FileTimeIndex;
 import org.apache.iotdb.db.storageengine.dataregion.tsfile.timeindex.ITimeIndex;
 import org.apache.iotdb.db.storageengine.dataregion.utils.TsFileResourceUtils;
+import org.apache.iotdb.db.storageengine.load.converter.LoadTsFileDataTypeConverter;
 import org.apache.iotdb.db.storageengine.load.memory.LoadTsFileAnalyzeSchemaMemoryBlock;
 import org.apache.iotdb.db.storageengine.load.memory.LoadTsFileMemoryManager;
 import org.apache.iotdb.db.utils.ModificationUtils;
@@ -129,6 +131,8 @@ public class LoadTsFileAnalyzer implements AutoCloseable {
 
   private final SchemaAutoCreatorAndVerifier schemaAutoCreatorAndVerifier;
 
+  private final LoadTsFileDataTypeConverter loadTsFileDataTypeConverter;
+
   LoadTsFileAnalyzer(
       LoadTsFileStatement loadTsFileStatement,
       MPPQueryContext context,
@@ -141,6 +145,7 @@ public class LoadTsFileAnalyzer implements AutoCloseable {
     this.schemaFetcher = schemaFetcher;
 
     this.schemaAutoCreatorAndVerifier = new SchemaAutoCreatorAndVerifier();
+    this.loadTsFileDataTypeConverter = new LoadTsFileDataTypeConverter();
   }
 
   public Analysis analyzeFileByFile(final boolean isDeleteAfterLoad) {
@@ -179,6 +184,11 @@ public Analysis analyzeFileByFile(final boolean isDeleteAfterLoad) {
         }
       } catch (AuthException e) {
         return createFailAnalysisForAuthException(e);
+      } catch (VerifyMetadataTypeMismatchException e) {
+        executeDataTypeConversionOnTypeMismatch(analysis, e);
+        // just return false to STOP the analysis process,
+        // the real result on the conversion will be set in the analysis.
+        return analysis;
       } catch (Exception e) {
         final String exceptionMessage =
             String.format(
@@ -195,6 +205,11 @@ public Analysis analyzeFileByFile(final boolean isDeleteAfterLoad) {
       schemaAutoCreatorAndVerifier.flush();
     } catch (AuthException e) {
       return createFailAnalysisForAuthException(e);
+    } catch (VerifyMetadataTypeMismatchException e) {
+      executeDataTypeConversionOnTypeMismatch(analysis, e);
+      // just return false to STOP the analysis process,
+      // the real result on the conversion will be set in the analysis.
+      return analysis;
     } catch (Exception e) {
       final String exceptionMessage =
           String.format(
@@ -214,13 +229,33 @@ public Analysis analyzeFileByFile(final boolean isDeleteAfterLoad) {
     return analysis;
   }
 
+  private void executeDataTypeConversionOnTypeMismatch(
+      final Analysis analysis, final VerifyMetadataTypeMismatchException e) {
+    final TSStatus status =
+        loadTsFileStatement.isConvertOnTypeMismatch()
+            ? loadTsFileDataTypeConverter.convertForTreeModel(loadTsFileStatement)
+            : null;
+
+    if (status == null) {
+      analysis.setFailStatus(
+          new TSStatus(TSStatusCode.LOAD_FILE_ERROR.getStatusCode()).setMessage(e.getMessage()));
+    } else if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()
+        && status.getCode() != TSStatusCode.REDIRECTION_RECOMMEND.getStatusCode()
+        && status.getCode() != TSStatusCode.LOAD_IDEMPOTENT_CONFLICT_EXCEPTION.getStatusCode()) {
+      analysis.setFailStatus(status);
+    }
+
+    analysis.setFinishQueryAfterAnalyze(true);
+    analysis.setStatement(loadTsFileStatement);
+  }
+
   @Override
   public void close() {
     schemaAutoCreatorAndVerifier.close();
   }
 
   private void analyzeSingleTsFile(final File tsFile, final boolean isDeleteAfterLoad)
-      throws IOException, AuthException {
+      throws IOException, AuthException, VerifyMetadataTypeMismatchException {
     try (final TsFileSequenceReader reader = new TsFileSequenceReader(tsFile.getAbsolutePath())) {
       // can be reused when constructing tsfile resource
       final TsFileSequenceReaderTimeseriesMetadataIterator timeseriesMetadataIterator =
@@ -310,7 +345,7 @@ public void setCurrentModificationsAndTimeIndex(TsFileResource resource) throws
     public void autoCreateAndVerify(
         TsFileSequenceReader reader,
         Map<IDeviceID, List<TimeseriesMetadata>> device2TimeseriesMetadataList)
-        throws IOException, AuthException {
+        throws IOException, AuthException, VerifyMetadataTypeMismatchException {
       for (final Map.Entry<IDeviceID, List<TimeseriesMetadata>> entry :
           device2TimeseriesMetadataList.entrySet()) {
         final IDeviceID device = entry.getKey();
@@ -412,13 +447,14 @@ public void flushAndClearDeviceIsAlignedCacheIfNecessary() throws SemanticExcept
       schemaCache.clearDeviceIsAlignedCacheIfNecessary();
     }
 
-    public void flush() throws AuthException {
+    public void flush() throws AuthException, VerifyMetadataTypeMismatchException {
       doAutoCreateAndVerify();
 
       schemaCache.clearTimeSeries();
     }
 
-    private void doAutoCreateAndVerify() throws SemanticException, AuthException {
+    private void doAutoCreateAndVerify()
+        throws SemanticException, AuthException, VerifyMetadataTypeMismatchException {
       if (schemaCache.getDevice2TimeSeries().isEmpty()) {
         return;
       }
@@ -439,7 +475,7 @@ private void doAutoCreateAndVerify() throws SemanticException, AuthException {
         if (loadTsFileStatement.isVerifySchema()) {
           verifySchema(schemaTree);
         }
-      } catch (AuthException e) {
+      } catch (AuthException | VerifyMetadataTypeMismatchException e) {
         throw e;
       } catch (Exception e) {
         LOGGER.warn("Auto create or verify schema error.", e);
@@ -600,7 +636,7 @@ private ISchemaTree autoCreateSchema() throws IllegalPathException {
     }
 
     private void verifySchema(ISchemaTree schemaTree)
-        throws VerifyMetadataException, IllegalPathException {
+        throws VerifyMetadataException, IllegalPathException, VerifyMetadataTypeMismatchException {
       for (final Map.Entry<IDeviceID, Set<MeasurementSchema>> entry :
           schemaCache.getDevice2TimeSeries().entrySet()) {
         final IDeviceID device = entry.getKey();
@@ -648,7 +684,7 @@ private void verifySchema(ISchemaTree schemaTree)
 
           // check datatype
           if (!tsFileSchema.getType().equals(iotdbSchema.getType())) {
-            throw new VerifyMetadataException(
+            throw new VerifyMetadataTypeMismatchException(
                 String.format(
                     "Measurement %s%s%s datatype not match, TsFile: %s, IoTDB: %s",
                     device,