Load: Added "skipFailedTableSchemaCheck" parameter (apache#16522)

* fix

fix

* tested

Author: Caideyipi

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

@@ -21,6 +21,7 @@
 
 import org.apache.iotdb.commons.exception.IllegalPathException;
 import org.apache.iotdb.commons.path.PatternTreeMap;
+import org.apache.iotdb.commons.pipe.config.PipeConfig;
 import org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory;
 import org.apache.iotdb.confignode.rpc.thrift.TDatabaseSchema;
 import org.apache.iotdb.db.auth.AuthorityChecker;
@@ -146,7 +147,18 @@ public void autoCreateAndVerify(final IDeviceID device) throws LoadAnalyzeExcept
           e);
     }
 
-    createTableAndDatabaseIfNecessary(device.getTableName());
+    try {
+      createTableAndDatabaseIfNecessary(device.getTableName());
+    } catch (final Exception e) {
+      if (PipeConfig.getInstance().isSkipFailedTableSchemaCheck()) {
+        LOGGER.info(
+            "Failed to check table schema, will skip because skipFailedTableSchemaCheck is set to true, message: {}",
+            e.getMessage());
+      } else {
+        throw e;
+      }
+    }
+
     // TODO: add permission check and record auth cost
     addDevice(device);
     if (shouldFlushDevices()) {

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/TableHeaderSchemaValidator.java

@@ -132,37 +132,37 @@ public Optional<TableSchema> validateTableHeaderSchema(
       DataNodeTreeViewSchemaUtils.checkTableInWrite(database, table);
       // If table with this name already exists and isStrictTagColumn is true, make sure the
       // existing
-      // id columns are the prefix of the incoming id columns, or vice versa
+      // id columns are the prefix of the incoming tag columns, or vice versa
       if (isStrictTagColumn) {
         final List<TsTableColumnSchema> realTagColumns = table.getTagColumnSchemaList();
         final List<ColumnSchema> incomingTagColumns = tableSchema.getTagColumns();
         if (realTagColumns.size() <= incomingTagColumns.size()) {
-          // When incoming table has more ID columns, the existing id columns
-          // should be the prefix of the incoming id columns (or equal)
+          // When incoming table has more TAG columns, the existing tag columns
+          // should be the prefix of the incoming tag columns (or equal)
           for (int indexReal = 0; indexReal < realTagColumns.size(); indexReal++) {
             final String tagName = realTagColumns.get(indexReal).getColumnName();
             final int indexIncoming = tableSchema.getIndexAmongTagColumns(tagName);
             if (indexIncoming != indexReal) {
               throw new LoadAnalyzeTableColumnDisorderException(
                   String.format(
-                      "Can not create table because incoming table has no less id columns than existing table, "
-                          + "and the existing id columns are not the prefix of the incoming id columns. "
-                          + "Existing id column: %s, index in existing table: %s, index in incoming table: %s",
+                      "Can not create table because incoming table has no less tag columns than existing table, "
+                          + "and the existing tag columns are not the prefix of the incoming tag columns. "
+                          + "Existing tag column: %s, index in existing table: %s, index in incoming table: %s",
                       tagName, indexReal, indexIncoming));
             }
           }
         } else {
-          // When existing table has more ID columns, the incoming id columns
-          // should be the prefix of the existing id columns
+          // When existing table has more TAG columns, the incoming tag columns
+          // should be the prefix of the existing tag columns
           for (int indexIncoming = 0; indexIncoming < incomingTagColumns.size(); indexIncoming++) {
             final String tagName = incomingTagColumns.get(indexIncoming).getName();
             final int indexReal = table.getTagColumnOrdinal(tagName);
             if (indexReal != indexIncoming) {
               throw new LoadAnalyzeTableColumnDisorderException(
                   String.format(
-                      "Can not create table because existing table has more id columns than incoming table, "
-                          + "and the incoming id columns are not the prefix of the existing id columns. "
-                          + "Incoming id column: %s, index in existing table: %s, index in incoming table: %s",
+                      "Can not create table because existing table has more tag columns than incoming table, "
+                          + "and the incoming tag columns are not the prefix of the existing tag columns. "
+                          + "Incoming tag column: %s, index in existing table: %s, index in incoming table: %s",
                       tagName, indexReal, indexIncoming));
             }
           }

iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java

@@ -209,6 +209,7 @@ public class CommonConfig {
   // Sequentially poll the tsFile by default
   private int pipeRealTimeQueuePollHistoricalTsFileThreshold = 1;
   private int pipeRealTimeQueueMaxWaitingTsFileSize = 1;
+  private boolean skipFailedTableSchemaCheck = true;
 
   /** The maximum number of threads that can be used to execute subtasks in PipeSubtaskExecutor. */
   private int pipeSubtaskExecutorMaxThreadNum =
@@ -1522,6 +1523,18 @@ public void setPipeRealTimeQueueMaxWaitingTsFileSize(int pipeRealTimeQueueMaxWai
         pipeRealTimeQueueMaxWaitingTsFileSize);
   }
 
+  public boolean isSkipFailedTableSchemaCheck() {
+    return skipFailedTableSchemaCheck;
+  }
+
+  public void setSkipFailedTableSchemaCheck(boolean skipFailedTableSchemaCheck) {
+    if (this.skipFailedTableSchemaCheck == skipFailedTableSchemaCheck) {
+      return;
+    }
+    this.skipFailedTableSchemaCheck = skipFailedTableSchemaCheck;
+    logger.info("skipFailedTableSchemaCheck is set to {}.", skipFailedTableSchemaCheck);
+  }
+
   public void setPipeAirGapReceiverEnabled(boolean pipeAirGapReceiverEnabled) {
     if (pipeAirGapReceiverEnabled == this.pipeAirGapReceiverEnabled) {
       return;

iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeConfig.java

@@ -119,6 +119,10 @@ public int getPipeRealTimeQueueMaxWaitingTsFileSize() {
     return COMMON_CONFIG.getPipeRealTimeQueueMaxWaitingTsFileSize();
   }
 
+  public boolean isSkipFailedTableSchemaCheck() {
+    return COMMON_CONFIG.isSkipFailedTableSchemaCheck();
+  }
+
   /////////////////////////////// Subtask Executor ///////////////////////////////
 
   public int getPipeSubtaskExecutorMaxThreadNum() {

iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeDescriptor.java

@@ -278,6 +278,11 @@ public static void loadPipeInternalConfig(CommonConfig config, TrimProperties pr
             properties.getProperty(
                 "pipe_realTime_queue_max_waiting_tsFile_size",
                 String.valueOf(config.getPipeRealTimeQueueMaxWaitingTsFileSize()))));
+    config.setSkipFailedTableSchemaCheck(
+        Boolean.parseBoolean(
+            properties.getProperty(
+                "skip_failed_table_schema_check",
+                String.valueOf(config.isSkipFailedTableSchemaCheck()))));
     config.setPipeSubtaskExecutorBasicCheckPointIntervalByConsumedEventCount(
         Integer.parseInt(
             properties.getProperty(