Load: Support auto data type conversion when data type mismatch detec…

…ted during analysis stage (#14529) Co-authored-by: Steve Yurong Su <[email protected]>
apache · Dec 31, 2024 · dd4d2be · dd4d2be
1 parent f7dcbc2
commit dd4d2be
Show file tree

Hide file tree

Showing 20 changed files with 832 additions and 69 deletions.
diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBLoadTsFileIT.java b/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";

diff --git a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TSStatusCode.java b/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),

diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/exception/VerifyMetadataException.java b/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);
+  }
 }
diff --git a/...node/src/main/java/org/apache/iotdb/db/exception/VerifyMetadataTypeMismatchException.java b/...node/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());
+  }
+}
diff --git a/.../src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/legacy/loader/TsFileLoader.java b/.../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);

diff --git a/...rc/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java b/...rc/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java
@@ -569,6 +569,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);
 

diff --git a/...e/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileAnalyzer.java b/...e/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileAnalyzer.java
@@ -19,9 +19,11 @@
 
 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.VerifyMetadataException;
+import org.apache.iotdb.db.exception.VerifyMetadataTypeMismatchException;
 import org.apache.iotdb.db.exception.load.LoadReadOnlyException;
 import org.apache.iotdb.db.exception.sql.SemanticException;
 import org.apache.iotdb.db.queryengine.common.MPPQueryContext;
@@ -33,6 +35,7 @@
 import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.LoadTsFile;
 import org.apache.iotdb.db.queryengine.plan.statement.crud.LoadTsFileStatement;
 import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
+import org.apache.iotdb.db.storageengine.load.converter.LoadTsFileDataTypeConverter;
 import org.apache.iotdb.rpc.RpcUtils;
 import org.apache.iotdb.rpc.TSStatusCode;
 
@@ -54,8 +57,7 @@ public abstract class LoadTsFileAnalyzer implements AutoCloseable {
   private static final Logger LOGGER = LoggerFactory.getLogger(LoadTsFileAnalyzer.class);
 
   // These are only used when constructed from tree model SQL
-  private final LoadTsFileStatement loadTsFileStatement;
-
+  private final LoadTsFileStatement loadTsFileTreeStatement;
   // These are only used when constructed from table model SQL
   private final LoadTsFile loadTsFileTableStatement;
 
@@ -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 LoadTsFileDataTypeConverter loadTsFileDataTypeConverter;
+
   LoadTsFileAnalyzer(LoadTsFileStatement loadTsFileStatement, MPPQueryContext context) {
-    this.loadTsFileStatement = loadTsFileStatement;
+    this.loadTsFileTreeStatement = 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.loadTsFileDataTypeConverter = new LoadTsFileDataTypeConverter();
 
     this.loadTsFileTableStatement = null;
     this.isTableModelStatement = false;
@@ -99,11 +107,13 @@ 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.loadTsFileDataTypeConverter = new LoadTsFileDataTypeConverter();
 
-    this.loadTsFileStatement = null;
+    this.loadTsFileTreeStatement = null;
     this.isTableModelStatement = true;
     this.context = context;
   }
@@ -137,6 +147,11 @@ protected boolean doAnalyzeFileByFile(IAnalysis analysis) {
       } catch (AuthException e) {
         setFailAnalysisForAuthException(analysis, e);
         return false;
+      } 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 false;
       } catch (BufferUnderflowException e) {
         LOGGER.warn(
             "The file {} is not a valid tsfile. Please check the input file.", tsFile.getPath(), e);
@@ -161,6 +176,39 @@ protected boolean doAnalyzeFileByFile(IAnalysis analysis) {
   protected abstract void analyzeSingleTsFile(final File tsFile)
       throws IOException, AuthException, VerifyMetadataException;
 
+  protected void executeDataTypeConversionOnTypeMismatch(
+      final IAnalysis analysis, final VerifyMetadataTypeMismatchException e) {
+    final TSStatus status =
+        isConvertOnTypeMismatch
+            ? (isTableModelStatement
+                ? loadTsFileDataTypeConverter.convertForTableModel(loadTsFileTableStatement)
+                : loadTsFileDataTypeConverter.convertForTreeModel(loadTsFileTreeStatement))
+            : 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);
+    setRealStatement(analysis);
+  }
+
+  protected void setRealStatement(IAnalysis analysis) {
+    if (isTableModelStatement) {
+      // Do nothing by now.
+    } else {
+      analysis.setRealStatement(loadTsFileTreeStatement);
+    }
+  }
+
+  protected String getStatementString() {
+    return statementString;
+  }
+
   protected TsFileResource constructTsFileResource(
       final TsFileSequenceReader reader, final File tsFile) throws IOException {
     final TsFileResource tsFileResource = new TsFileResource(tsFile);
@@ -174,38 +222,30 @@ protected TsFileResource constructTsFileResource(
     return tsFileResource;
   }
 
-  protected String getStatementString() {
-    return statementString;
-  }
-
-  protected void setRealStatement(IAnalysis analysis) {
-    if (isTableModelStatement) {
-      // Do nothing by now.
-    } else {
-      analysis.setRealStatement(loadTsFileStatement);
-    }
-  }
-
   protected void addTsFileResource(TsFileResource tsFileResource) {
     if (isTableModelStatement) {
       loadTsFileTableStatement.addTsFileResource(tsFileResource);
     } else {
-      loadTsFileStatement.addTsFileResource(tsFileResource);
+      loadTsFileTreeStatement.addTsFileResource(tsFileResource);
     }
   }
 
   protected void addWritePointCount(long writePointCount) {
     if (isTableModelStatement) {
       loadTsFileTableStatement.addWritePointCount(writePointCount);
     } else {
-      loadTsFileStatement.addWritePointCount(writePointCount);
+      loadTsFileTreeStatement.addWritePointCount(writePointCount);
     }
   }
 
   protected boolean isVerifySchema() {
     return isVerifySchema;
   }
 
+  protected boolean isConvertOnTypeMismatch() {
+    return isConvertOnTypeMismatch;
+  }
+
   protected boolean isAutoCreateDatabase() {
     return isAutoCreateDatabase;
   }

diff --git a/...in/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileTableSchemaCache.java b/...in/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileTableSchemaCache.java
@@ -24,6 +24,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.LoadRuntimeOutOfMemoryException;
 import org.apache.iotdb.db.exception.sql.SemanticException;
 import org.apache.iotdb.db.queryengine.common.MPPQueryContext;
@@ -290,7 +291,7 @@ private void verifyTableDataTypeAndGenerateIdColumnMapper(
         final ColumnSchema realColumn =
             realSchema.getColumn(fileColumn.getName(), fileColumn.getColumnCategory());
         if (!fileColumn.getType().equals(realColumn.getType())) {
-          throw new VerifyMetadataException(
+          throw new VerifyMetadataTypeMismatchException(
               String.format(
                   "Data type mismatch for column %s in table %s, type in TsFile: %s, type in IoTDB: %s",
                   realColumn.getName(),