Load: Support auto data type conversion during Analysis stage

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.IMeasurementSchema;
 import org.apache.tsfile.write.schema.MeasurementSchema;
 import org.junit.After;
@@ -48,6 +49,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;
@@ -56,6 +58,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;
@@ -328,7 +331,7 @@ public void testLoadWithExtendTemplate() throws Exception {
   }
 
   @Test
-  public void testLoadWithAutoRegister() throws Exception {
+  public void testLoadWithAutoCreate() throws Exception {
     final long writtenPoint1;
     // device 0, device 1, sg 0
     try (final TsFileGenerator generator =
@@ -898,6 +901,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<IMeasurementSchema> 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

@@ -197,6 +197,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/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/IdentifierParser.g4

@@ -69,6 +69,7 @@ keyWords
     | CONTAIN
     | CONTAINS
     | CONTINUOUS
+    | CONVERTONTYPEMISMATCH
     | COUNT
     | CQ
     | CQS

iotdb-core/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/IoTDBSqlParser.g4

@@ -1141,6 +1141,7 @@ loadFileAttributeClause
     : SGLEVEL operator_eq INTEGER_LITERAL
     | VERIFY operator_eq boolean_literal
     | ONSUCCESS operator_eq (DELETE|NONE)
+    | CONVERTONTYPEMISMATCH operator_eq boolean_literal
     ;
 
 loadFileWithAttributeClauses

iotdb-core/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/SqlLexer.g4

@@ -189,6 +189,10 @@ CONTINUOUS
     : C O N T I N U O U S
     ;
 
+CONVERTONTYPEMISMATCH
+    : C O N V E R T O N T Y P E M I S M A T C H
+    ;
+
 COUNT
     : C O U N T
     ;

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

@@ -571,6 +571,7 @@ private TSStatus loadTsFileSync(final String dataBaseName, final String fileAbso
       throws FileNotFoundException {
     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/load/LoadTsFileAnalyzer.java

@@ -19,10 +19,12 @@
 
 package org.apache.iotdb.db.queryengine.plan.analyze.load;
 
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
 import org.apache.iotdb.commons.auth.AuthException;
 import org.apache.iotdb.commons.conf.CommonDescriptor;
 import org.apache.iotdb.db.exception.LoadReadOnlyException;
 import org.apache.iotdb.db.exception.VerifyMetadataException;
+import org.apache.iotdb.db.exception.VerifyMetadataTypeMismatchException;
 import org.apache.iotdb.db.exception.sql.SemanticException;
 import org.apache.iotdb.db.queryengine.common.MPPQueryContext;
 import org.apache.iotdb.db.queryengine.plan.analyze.ClusterPartitionFetcher;
@@ -67,6 +69,8 @@ public abstract class LoadTsFileAnalyzer implements AutoCloseable {
 
   protected final boolean isDeleteAfterLoad;
 
+  protected final boolean isConvertOnTypeMismatch;
+
   protected final boolean isAutoCreateDatabase;
 
   protected final int databaseLevel;
@@ -78,15 +82,19 @@ public abstract class LoadTsFileAnalyzer implements AutoCloseable {
   final IPartitionFetcher partitionFetcher = ClusterPartitionFetcher.getInstance();
   final ISchemaFetcher schemaFetcher = ClusterSchemaFetcher.getInstance();
 
+  protected final LoadTsFileDataTypeMismatchConvertHandler loadTsFileDataTypeMismatchConvertHandler;
+
   LoadTsFileAnalyzer(LoadTsFileStatement loadTsFileStatement, MPPQueryContext context) {
     this.loadTsFileStatement = loadTsFileStatement;
     this.tsFiles = loadTsFileStatement.getTsFiles();
     this.statementString = loadTsFileStatement.toString();
     this.isVerifySchema = loadTsFileStatement.isVerifySchema();
     this.isDeleteAfterLoad = loadTsFileStatement.isDeleteAfterLoad();
+    this.isConvertOnTypeMismatch = loadTsFileStatement.isConvertOnTypeMismatch();
     this.isAutoCreateDatabase = loadTsFileStatement.isAutoCreateDatabase();
     this.databaseLevel = loadTsFileStatement.getDatabaseLevel();
     this.database = loadTsFileStatement.getDatabase();
+    this.loadTsFileDataTypeMismatchConvertHandler = new LoadTsFileDataTypeMismatchConvertHandler();
 
     this.loadTsFileTableStatement = null;
     this.isTableModelStatement = false;
@@ -99,9 +107,11 @@ public abstract class LoadTsFileAnalyzer implements AutoCloseable {
     this.statementString = loadTsFileTableStatement.toString();
     this.isVerifySchema = true;
     this.isDeleteAfterLoad = loadTsFileTableStatement.isDeleteAfterLoad();
+    this.isConvertOnTypeMismatch = loadTsFileTableStatement.isConvertOnTypeMismatch();
     this.isAutoCreateDatabase = loadTsFileTableStatement.isAutoCreateDatabase();
     this.databaseLevel = loadTsFileTableStatement.getDatabaseLevel();
     this.database = loadTsFileTableStatement.getDatabase();
+    this.loadTsFileDataTypeMismatchConvertHandler = new LoadTsFileDataTypeMismatchConvertHandler();
 
     this.loadTsFileStatement = null;
     this.isTableModelStatement = true;
@@ -137,6 +147,11 @@ protected boolean doAnalyzeFileByFile(IAnalysis analysis) {
       } catch (AuthException e) {
         setFailAnalysisForAuthException(analysis, e);
         return false;
+      } catch (VerifyMetadataTypeMismatchException e) {
+        this.executeDataTypeConversionOnTypeMismatch(analysis);
+        // just return false to STOP the analysis process, the real result on the conversion will be
+        // set in the analysis.
+        return false;
       } catch (BufferUnderflowException e) {
         LOGGER.warn(
             "The file {} is not a valid tsfile. Please check the input file.", tsFile.getPath(), e);
@@ -174,6 +189,24 @@ protected TsFileResource constructTsFileResource(
     return tsFileResource;
   }
 
+  public void executeDataTypeConversionOnTypeMismatch(IAnalysis analysis) {
+
+    TSStatus status;
+    if (isTableModelStatement) {
+      status =
+          loadTsFileDataTypeMismatchConvertHandler.convertForTableModel(loadTsFileTableStatement);
+    } else {
+      status = loadTsFileDataTypeMismatchConvertHandler.convertForTreeModel(loadTsFileStatement);
+    }
+
+    if (status == null || status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+      analysis.setFailStatus(status);
+    }
+
+    analysis.setFinishQueryAfterAnalyze(true);
+    setRealStatement(analysis);
+  }
+
   protected String getStatementString() {
     return statementString;
   }
@@ -206,6 +239,10 @@ protected boolean isVerifySchema() {
     return isVerifySchema;
   }
 
+  protected boolean isConvertOnTypeMismatch() {
+    return isConvertOnTypeMismatch;
+  }
+
   protected boolean isAutoCreateDatabase() {
     return isAutoCreateDatabase;
   }