Honor auto-create-schema settings for tree and table LOAD

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

@@ -22,6 +22,7 @@
 import org.apache.iotdb.commons.auth.entity.PrivilegeType;
 import org.apache.iotdb.commons.path.MeasurementPath;
 import org.apache.iotdb.commons.schema.column.ColumnHeaderConstant;
+import org.apache.iotdb.db.i18n.StorageEngineMessages;
 import org.apache.iotdb.db.it.utils.TestUtils;
 import org.apache.iotdb.db.storageengine.dataregion.modification.ModificationFile;
 import org.apache.iotdb.db.storageengine.dataregion.modification.TreeDeletionEntry;
@@ -34,13 +35,19 @@
 import org.apache.iotdb.itbase.category.ClusterIT;
 import org.apache.iotdb.itbase.category.LocalStandaloneIT;
 import org.apache.iotdb.jdbc.IoTDBSQLException;
+import org.apache.iotdb.rpc.TSStatusCode;
 
 import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.file.metadata.IDeviceID;
 import org.apache.tsfile.file.metadata.enums.TSEncoding;
+import org.apache.tsfile.read.TsFileSequenceReader;
 import org.apache.tsfile.read.common.Path;
 import org.apache.tsfile.utils.Pair;
+import org.apache.tsfile.write.TsFileWriter;
+import org.apache.tsfile.write.record.Tablet;
 import org.apache.tsfile.write.schema.IMeasurementSchema;
 import org.apache.tsfile.write.schema.MeasurementSchema;
+import org.awaitility.Awaitility;
 import org.junit.After;
 import org.junit.Assert;
 import org.junit.Before;
@@ -52,6 +59,8 @@
 import org.slf4j.LoggerFactory;
 
 import java.io.File;
+import java.io.RandomAccessFile;
+import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
 import java.sql.Connection;
 import java.sql.ResultSet;
@@ -186,6 +195,119 @@ private boolean deleteDir() {
     return tmpDir.delete();
   }
 
+  @Test
+  public void testQueryFallbackHonorsDisabledSchemaCreation() throws Exception {
+    final String database = "root.load.fallback_disabled_schema";
+    final File source = new File(tmpDir, "712-0-0-0.tsfile");
+    final long[] times = {-10001, -10000, -1, 0, 9999, 10000};
+    final List<IMeasurementSchema> schemas =
+        Arrays.asList(
+            new MeasurementSchema("s", TSDataType.INT64),
+            new MeasurementSchema("txt", TSDataType.TEXT));
+    try (TsFileWriter writer = new TsFileWriter(source)) {
+      for (int device = 0; device < 2; device++) {
+        final String devicePath = database + ".d" + device;
+        writer.registerAlignedTimeseries(devicePath, schemas);
+        final Tablet tablet = new Tablet(devicePath, schemas, times.length);
+        for (int row = 0; row < times.length; row++) {
+          tablet.addTimestamp(row, times[row]);
+          tablet.addValue(row, "s", device * 100L + row);
+          tablet.addValue(row, "txt", row == 0 ? null : "d" + device + "_r" + row);
+        }
+        writer.writeAligned(tablet);
+      }
+    }
+    final TsFileResource resource = new TsFileResource(source);
+    for (int device = 0; device < 2; device++) {
+      final IDeviceID id = IDeviceID.Factory.DEFAULT_FACTORY.create(database + ".d" + device);
+      resource.updateStartTime(id, times[0]);
+      resource.updateEndTime(id, times[times.length - 1]);
+    }
+    resource.serialize();
+    try (ModificationFile mods = resource.getExclusiveModFile()) {
+      for (int device = 0; device < 2; device++) {
+        mods.write(
+            new TreeDeletionEntry(
+                new MeasurementPath(database + ".d" + device + ".s"), times[1], times[1]));
+      }
+    }
+    try (TsFileSequenceReader reader = new TsFileSequenceReader(source.getAbsolutePath());
+        RandomAccessFile bytes = new RandomAccessFile(source, "rw")) {
+      final long chunk =
+          reader
+              .getIChunkMetadataList(
+                  IDeviceID.Factory.DEFAULT_FACTORY.create(database + ".d1"), "s")
+              .get(0)
+              .getOffsetOfChunkHeader();
+      reader.position(chunk);
+      reader.readChunkHeader(reader.readMarker());
+      // Preserve readable metadata, but make both the piece reader and scan parser reject this
+      // chunk. Multiple time partitions prevent LocalLoadStrategy from bypassing the piece reader.
+      bytes.seek(reader.position() - 2);
+      bytes.writeByte(0x7f);
+    }
+    final Map<File, byte[]> sourceFiles = new HashMap<>();
+    for (String suffix :
+        new String[] {"", TsFileResource.RESOURCE_SUFFIX, ModificationFile.FILE_SUFFIX}) {
+      final File file = new File(source + suffix);
+      sourceFiles.put(file, Files.readAllBytes(file.toPath()));
+    }
+    final DataNodeWrapper node = EnvFactory.getEnv().getDataNodeWrapperList().get(0);
+    final File log = new File(node.getLogDirPath(), node.getId() + ".log");
+    final long logOffset = Files.size(log.toPath());
+    try (Connection connection = EnvFactory.getEnv().getConnectionWithSpecifiedDataNode(node);
+        Statement statement = connection.createStatement()) {
+      statement.execute("create database " + database);
+      try {
+        final SQLException failure =
+            Assert.assertThrows(
+                SQLException.class,
+                () ->
+                    statement.execute(
+                        "load '"
+                            + source.getAbsolutePath()
+                            + "' with ('database-level'='2','verify'='false',"
+                            + "'auto-create-schema'='false','tablet-conversion-threshold'='-1',"
+                            + "'on-success'='delete')"));
+        Assert.assertEquals(TSStatusCode.LOAD_FILE_ERROR.getStatusCode(), failure.getErrorCode());
+        Assert.assertTrue(failure.getMessage(), failure.getMessage().contains(source.getName()));
+        final String fallbackMessage =
+            String.format(
+                StorageEngineMessages
+                    .MESSAGE_LOAD_SCAN_PARSER_DETECTED_A_CORRUPTED_SECTION_IN_TSFILE_ARG_AT_DEVICE_ARG_SWITCH_TO_QUERY_PARSING_FOR_REMAINING_DEVICES_EFA07985
+                    .replace("{}", "%s"),
+                source.getAbsolutePath(),
+                database + ".d1");
+        Awaitility.await()
+            .atMost(30, TimeUnit.SECONDS)
+            .untilAsserted(
+                () -> {
+                  try (RandomAccessFile recentLog = new RandomAccessFile(log, "r")) {
+                    Assert.assertTrue(
+                        "The node log must not be truncated", recentLog.length() >= logOffset);
+                    recentLog.seek(logOffset);
+                    final byte[] bytes = new byte[Math.toIntExact(recentLog.length() - logOffset)];
+                    recentLog.readFully(bytes);
+                    Assert.assertTrue(
+                        "This LOAD must reach Scan-to-Query fallback for its own source file",
+                        new String(bytes, StandardCharsets.UTF_8).contains(fallbackMessage));
+                  }
+                });
+        try (ResultSet schema = statement.executeQuery("show timeseries " + database + ".**")) {
+          Assert.assertFalse("Schema must remain absent after query fallback", schema.next());
+        }
+        for (Map.Entry<File, byte[]> entry : sourceFiles.entrySet()) {
+          Assert.assertArrayEquals(
+              "Failed LOAD must preserve " + entry.getKey(),
+              entry.getValue(),
+              Files.readAllBytes(entry.getKey().toPath()));
+        }
+      } finally {
+        statement.execute("delete database " + database);
+      }
+    }
+  }
+
   @Test
   public void testLoad() throws Exception {
     registerSchema();

iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java

@@ -523,6 +523,9 @@ public final class DataNodeQueryMessages {
       "Auto create or verify schema error.";
   public static final String LOAD_TSFILE_DEVICE_SCHEMA_MISSING_AUTO_CREATE_DISABLED =
       "Device %s does not exist in IoTDB and can not be created. Please check whether auto-create-schema is enabled.";
+  public static final String
+      EXCEPTION_THE_DATABASE_ARG_DOES_NOT_EXIST_AND_AUTO_CREATE_SCHEMA_IS_DISABLED_28F8D49A =
+          "The database %s does not exist and auto-create-schema is disabled.";
   public static final String LOAD_TSFILE_MEASUREMENT_SCHEMA_MISSING_AUTO_CREATE_DISABLED =
       "Measurement %s does not exist in IoTDB and can not be created. Please check whether auto-create-schema is enabled.";
   public static final String PIPE_GENERATED_LOAD_TSFILE_WAITING_FOR_SCHEMA_METADATA =

iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java

@@ -503,6 +503,9 @@ public final class DataNodeQueryMessages {
       "自动创建或验证 schema 出错。";
   public static final String LOAD_TSFILE_DEVICE_SCHEMA_MISSING_AUTO_CREATE_DISABLED =
       "设备 %s 在 IoTDB 中不存在且无法被创建。请检查是否启用了 auto-create-schema。";
+  public static final String
+      EXCEPTION_THE_DATABASE_ARG_DOES_NOT_EXIST_AND_AUTO_CREATE_SCHEMA_IS_DISABLED_28F8D49A =
+          "数据库 %s 不存在，且 auto-create-schema 已禁用。";
   public static final String LOAD_TSFILE_MEASUREMENT_SCHEMA_MISSING_AUTO_CREATE_DISABLED =
       "时间序列 %s 在 IoTDB 中不存在且无法被创建。请检查是否启用了 auto-create-schema。";
   public static final String PIPE_GENERATED_LOAD_TSFILE_WAITING_FOR_SCHEMA_METADATA =

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

@@ -0,0 +1,27 @@
+/*
+ * 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.load;
+
+public class LoadAnalyzeMissingDatabaseException extends LoadAnalyzeMissingSchemaException {
+
+  public LoadAnalyzeMissingDatabaseException(final String message) {
+    super(message);
+  }
+}

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

@@ -31,6 +31,7 @@
 import org.apache.iotdb.db.conf.IoTDBDescriptor;
 import org.apache.iotdb.db.exception.load.LoadAnalyzeException;
 import org.apache.iotdb.db.exception.load.LoadAnalyzeInvalidPathException;
+import org.apache.iotdb.db.exception.load.LoadAnalyzeMissingDatabaseException;
 import org.apache.iotdb.db.exception.load.LoadAnalyzeMissingSchemaException;
 import org.apache.iotdb.db.exception.load.LoadAnalyzeTypeMismatchException;
 import org.apache.iotdb.db.exception.load.LoadEmptyFileException;
@@ -83,6 +84,7 @@
 import java.util.Optional;
 
 import static org.apache.iotdb.db.queryengine.plan.execution.config.TableConfigTaskVisitor.DATABASE_NOT_SPECIFIED;
+import static org.apache.iotdb.db.storageengine.load.config.LoadTsFileConfigurator.AUTO_CREATE_SCHEMA_KEY;
 import static org.apache.iotdb.db.storageengine.load.metrics.LoadTsFileCostMetricsSet.ANALYSIS;
 import static org.apache.iotdb.db.storageengine.load.metrics.LoadTsFileCostMetricsSet.ANALYSIS_ASYNC_MOVE;
 
@@ -152,7 +154,10 @@
     this.isAutoCreateSchemaRequested = loadTsFileStatement.isAutoCreateSchema();
     this.isAutoCreateSchemaEnabled =
         IoTDBDescriptor.getInstance().getConfig().isAutoCreateSchemaEnabled()
-            && isAutoCreateSchemaRequested;
+            && isAutoCreateSchemaRequested
+            && (context == null
+                || context.getSession() == null
+                || context.getSession().isEnableAutoCreateSchema());
     this.isAutoCreateDatabase = loadTsFileStatement.isAutoCreateDatabase();
     this.isDeleteAfterLoad = loadTsFileStatement.isDeleteAfterLoad();
     this.isConvertOnTypeMismatch = loadTsFileStatement.isConvertOnTypeMismatch();
@@ -180,7 +185,10 @@
     this.isAutoCreateSchemaRequested = loadTsFileTableStatement.isAutoCreateSchema();
     this.isAutoCreateSchemaEnabled =
         IoTDBDescriptor.getInstance().getConfig().isAutoCreateSchemaEnabled()
-            && isAutoCreateSchemaRequested;
+            && isAutoCreateSchemaRequested
+            && (context == null
+                || context.getSession() == null
+                || context.getSession().isEnableAutoCreateSchema());
     this.isAutoCreateDatabase = loadTsFileTableStatement.isAutoCreateDatabase();
     this.isDeleteAfterLoad = loadTsFileTableStatement.isDeleteAfterLoad();
     this.isConvertOnTypeMismatch = loadTsFileTableStatement.isConvertOnTypeMismatch();
@@ -253,7 +261,7 @@
       final String exceptionMessage =
           String.format(
               DataNodeQueryMessages
                   .EXCEPTION_AUTO_CREATE_OR_VERIFY_SCHEMA_ERROR_WHEN_EXECUTING_STATEMENT_ARG_DETAIL_ARG_B6C9E02E,
               getStatementString(),
               e.getMessage() == null ? e.getClass().getName() : e.getMessage());
       LOGGER.warn(exceptionMessage, e);
@@ -318,13 +326,14 @@
       } else {
         databaseName = null;
       }
+      // The active loader uses a new session, so carry the effective policy with it.
       final Map<String, String> activeLoadAttributes =
           ActiveLoadPathHelper.buildAttributes(
               databaseName,
               databaseLevel,
               isConvertOnTypeMismatch,
               isVerifySchema,
-              isAutoCreateSchemaRequested,
+              isAutoCreateSchemaEnabled,
               tabletConversionThresholdBytes,
               isGeneratedByPipe,
               Objects.nonNull(context) && Objects.nonNull(context.getSession())
@@ -480,8 +489,10 @@
                 sessionInfo.getSessionId(),
                 sessionInfo.getUserEntity(),
                 sessionInfo.getZoneId(),
+                sessionInfo.getVersion(),
                 sessionInfo.getDatabaseName().orElse(null),
-                SqlDialect.TABLE);
+                SqlDialect.TABLE,
+                sessionInfo.isEnableAutoCreateSchema());
         context.setSession(newSessionInfo);
         doAnalyzeSingleTableFile(tsFile, reader, timeseriesMetadataIterator, tableSchemaMap);
       } else {
@@ -490,8 +501,10 @@
                 sessionInfo.getSessionId(),
                 sessionInfo.getUserEntity(),
                 sessionInfo.getZoneId(),
+                sessionInfo.getVersion(),
                 sessionInfo.getDatabaseName().orElse(null),
-                SqlDialect.TREE);
+                SqlDialect.TREE,
+                sessionInfo.isEnableAutoCreateSchema());
         context.setSession(newSessionInfo);
         doAnalyzeSingleTreeFile(tsFile, reader, timeseriesMetadataIterator);
       }
@@ -545,23 +558,10 @@
       final TSStatus status =
           isTableModelTsFile.get(i)
               ? loadTsFileDataTypeConverter
-                  .convertForTableModel(
-                      (isGeneratedByPipe
-                              ? LoadTsFile.createForPipe(
-                                  null, tsFiles.get(i).getPath(), Collections.emptyMap())
-                              : LoadTsFile.createUnchecked(
-                                  null, tsFiles.get(i).getPath(), Collections.emptyMap()))
-                          .setDatabase(databaseForTableData)
-                          .setDeleteAfterLoad(isDeleteAfterLoad)
-                          .setConvertOnTypeMismatch(isConvertOnTypeMismatch))
+                  .convertForTableModel(createTableConversionStatement(tsFiles.get(i)))
                   .orElse(null)
               : loadTsFileDataTypeConverter
-                  .convertForTreeModel(
-                      (isGeneratedByPipe
-                              ? LoadTsFileStatement.createForPipe(tsFiles.get(i).getPath())
-                              : LoadTsFileStatement.createUnchecked(tsFiles.get(i).getPath()))
-                          .setDeleteAfterLoad(isDeleteAfterLoad)
-                          .setConvertOnTypeMismatch(isConvertOnTypeMismatch))
+                  .convertForTreeModel(createTreeConversionStatement(tsFiles.get(i)))
                   .orElse(null);
 
       if (status == null || !loadTsFileDataTypeConverter.isSuccessful(status)) {
@@ -728,7 +728,8 @@
   private LoadTsFileTableSchemaCache getOrCreateTableSchemaCache() {
     if (tableSchemaCache == null) {
       tableSchemaCache =
-          new LoadTsFileTableSchemaCache(metadata, context, isAutoCreateDatabase, isVerifySchema);
+          new LoadTsFileTableSchemaCache(
+              metadata, context, isAutoCreateDatabase, isVerifySchema, isAutoCreateSchemaEnabled);
     }
     return tableSchemaCache;
   }
@@ -849,23 +850,10 @@
         final TSStatus status =
             isTableModelTsFile.get(i)
                 ? loadTsFileDataTypeConverter
-                    .convertForTableModel(
-                        (isGeneratedByPipe
-                                ? LoadTsFile.createForPipe(
-                                    null, tsFiles.get(i).getPath(), Collections.emptyMap())
-                                : LoadTsFile.createUnchecked(
-                                    null, tsFiles.get(i).getPath(), Collections.emptyMap()))
-                            .setDatabase(databaseForTableData)
-                            .setDeleteAfterLoad(isDeleteAfterLoad)
-                            .setConvertOnTypeMismatch(isConvertOnTypeMismatch))
+                    .convertForTableModel(createTableConversionStatement(tsFiles.get(i)))
                     .orElse(null)
                 : loadTsFileDataTypeConverter
-                    .convertForTreeModel(
-                        (isGeneratedByPipe
-                                ? LoadTsFileStatement.createForPipe(tsFiles.get(i).getPath())
-                                : LoadTsFileStatement.createUnchecked(tsFiles.get(i).getPath()))
-                            .setDeleteAfterLoad(isDeleteAfterLoad)
-                            .setConvertOnTypeMismatch(isConvertOnTypeMismatch))
+                    .convertForTreeModel(createTreeConversionStatement(tsFiles.get(i)))
                     .orElse(null);
 
         if (status == null) {
@@ -911,6 +899,30 @@
     setRealStatement(analysis);
   }
 
+  private LoadTsFileStatement createTreeConversionStatement(final File tsFile)
+      throws FileNotFoundException {
+    final LoadTsFileStatement statement =
+        (isGeneratedByPipe
+                ? LoadTsFileStatement.createForPipe(tsFile.getPath())
+                : LoadTsFileStatement.createUnchecked(tsFile.getPath()))
+            .setDeleteAfterLoad(isDeleteAfterLoad)
+            .setConvertOnTypeMismatch(isConvertOnTypeMismatch);
+    statement.setAutoCreateSchema(isAutoCreateSchemaRequested);
+    return statement;
+  }
+
+  private LoadTsFile createTableConversionStatement(final File tsFile) {
+    final Map<String, String> attributes =
+        Collections.singletonMap(
+            AUTO_CREATE_SCHEMA_KEY, String.valueOf(isAutoCreateSchemaRequested));
+    return (isGeneratedByPipe
+            ? LoadTsFile.createForPipe(null, tsFile.getPath(), attributes)
+            : LoadTsFile.createUnchecked(null, tsFile.getPath(), attributes))
+        .setDatabase(databaseForTableData)
+        .setDeleteAfterLoad(isDeleteAfterLoad)
+        .setConvertOnTypeMismatch(isConvertOnTypeMismatch);
+  }
+
   private boolean setTemporaryUnavailableStatusIfNecessary(
       final IAnalysis analysis, final Throwable throwable) {
     // Region unavailability can persist beyond recovery. Retain the file for retry; permanent
@@ -955,6 +967,7 @@
 
   private boolean shouldSkipConversion(LoadAnalyzeException e) {
     return e instanceof LoadAnalyzeInvalidPathException
+        || e instanceof LoadAnalyzeMissingDatabaseException
         || (e instanceof LoadAnalyzeTypeMismatchException) && !isConvertOnTypeMismatch;
   }