[To dev/1.3] Honor auto-create-schema settings for tree LOAD

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

@@ -49,6 +49,7 @@
 import org.apache.tsfile.write.TsFileWriter;
 import org.apache.tsfile.write.record.Tablet;
 import org.apache.tsfile.write.schema.MeasurementSchema;
+import org.awaitility.Awaitility;
 import org.junit.After;
 import org.junit.Assert;
 import org.junit.Before;
@@ -62,6 +63,7 @@
 import java.io.File;
 import java.io.IOException;
 import java.io.RandomAccessFile;
+import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
 import java.sql.Connection;
 import java.sql.ResultSet;
@@ -202,6 +204,121 @@ 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<MeasurementSchema> 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(new Path(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("s", row, device * 100L + row);
+          tablet.addValue("txt", row, row == 0 ? null : "d" + device + "_r" + row);
+        }
+        tablet.rowSize = times.length;
+        writer.writeAligned(tablet);
+      }
+    }
+    final TsFileResource resource = new TsFileResource(source);
+    for (int device = 0; device < 2; device++) {
+      final IDeviceID id = new PlainDeviceID(database + ".d" + device);
+      resource.updateStartTime(id, times[0]);
+      resource.updateEndTime(id, times[times.length - 1]);
+    }
+    resource.serialize();
+    try (ModificationFile mods = resource.getModFile()) {
+      for (int device = 0; device < 2; device++) {
+        mods.write(
+            new Deletion(
+                new MeasurementPath(database + ".d" + device + ".s"),
+                Long.MAX_VALUE,
+                times[1],
+                times[1]));
+      }
+    }
+    try (TsFileSequenceReader reader = new TsFileSequenceReader(source.getAbsolutePath());
+        RandomAccessFile bytes = new RandomAccessFile(source, "rw")) {
+      final long chunk =
+          reader
+              .getChunkMetadataList(new Path(new PlainDeviceID(database + ".d1"), "s", false))
+              .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 local loading 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(
+                "Load: Scan parser detected a corrupted section in TsFile %s",
+                source.getAbsolutePath());
+        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);
+                    final String loadLog = new String(bytes, StandardCharsets.UTF_8);
+                    Assert.assertTrue(
+                        "This LOAD must reach Scan-to-Query fallback for its own source file",
+                        loadLog.contains(fallbackMessage)
+                            && loadLog.contains("Switch to query parsing for remaining devices."));
+                  }
+                });
+        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/pom.xml

@@ -341,6 +341,11 @@
             <artifactId>powermock-api-mockito2</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.powermock</groupId>
+            <artifactId>powermock-reflect</artifactId>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>org.awaitility</groupId>
             <artifactId>awaitility</artifactId>

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

@@ -64,6 +64,9 @@ private DataNodeQueryMessages() {}
           "There is not enough memory for Query %s, the contextHolder is %s,current remaining free "
               + "memory is %dB, already reserved memory for this context in total is %dB, the memory "
               + "requested this time is %dB";
+  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 LOG_DELETE_AFTER_LOADING_ARG_ERROR_C606A060 =
       "Delete After Loading {} error.";
 }

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

@@ -61,5 +61,8 @@ private DataNodeQueryMessages() {}
   public static final String
       QUERY_EXCEPTION_THERE_IS_NOT_ENOUGH_MEMORY_FOR_QUERY_S_THE_CONTEXTHOLDER_546CDD02 =
           "Query %s 内存不足，contextHolder 为 %s，当前剩余空闲内存为 %dB，该 context 已预留总内存为 %dB，本次请求内存为 %dB。";
+  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 LOG_DELETE_AFTER_LOADING_ARG_ERROR_C606A060 = "加载后删除 {} 时出错。";
 }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/exception/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;
+
+public class LoadAnalyzeMissingDatabaseException extends LoadAnalyzeMissingSchemaException {
+
+  public LoadAnalyzeMissingDatabaseException(final String message) {
+    super(message);
+  }
+}

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/common/SessionInfo.java

@@ -34,23 +34,33 @@ public class SessionInfo {
   private final String userName;
   private final ZoneId zoneId;
   private final String cliHostname;
+  // LOAD conversion uses this policy during local schema analysis; keep the session wire format.
+  private final boolean enableAutoCreateSchema;
 
   private ClientVersion version = ClientVersion.V_1_0;
 
   public SessionInfo(long sessionId, String userName, ZoneId zoneId, String cliHostname) {
-    this.sessionId = sessionId;
-    this.userName = userName;
-    this.zoneId = zoneId;
-    this.cliHostname = cliHostname;
+    this(sessionId, userName, zoneId, ClientVersion.V_1_0, cliHostname);
   }
 
   public SessionInfo(
       long sessionId, String userName, ZoneId zoneId, ClientVersion version, String cliHostname) {
+    this(sessionId, userName, zoneId, version, cliHostname, true);
+  }
+
+  public SessionInfo(
+      long sessionId,
+      String userName,
+      ZoneId zoneId,
+      ClientVersion version,
+      String cliHostname,
+      boolean enableAutoCreateSchema) {
     this.sessionId = sessionId;
     this.userName = userName;
     this.zoneId = zoneId;
     this.version = version;
     this.cliHostname = cliHostname;
+    this.enableAutoCreateSchema = enableAutoCreateSchema;
   }
 
   public long getSessionId() {
@@ -73,6 +83,10 @@ public String getCliHostname() {
     return cliHostname;
   }
 
+  public boolean isEnableAutoCreateSchema() {
+    return enableAutoCreateSchema;
+  }
+
   public static SessionInfo deserializeFrom(ByteBuffer buffer) {
     long sessionId = ReadWriteIOUtils.readLong(buffer);
     String userName = ReadWriteIOUtils.readString(buffer);

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

@@ -40,6 +40,7 @@
 import org.apache.iotdb.db.conf.IoTDBDescriptor;
 import org.apache.iotdb.db.exception.LoadAnalyzeException;
 import org.apache.iotdb.db.exception.LoadAnalyzeInvalidPathException;
+import org.apache.iotdb.db.exception.LoadAnalyzeMissingDatabaseException;
 import org.apache.iotdb.db.exception.LoadAnalyzeMissingSchemaException;
 import org.apache.iotdb.db.exception.LoadAnalyzeTypeMismatchException;
 import org.apache.iotdb.db.exception.load.LoadEmptyFileException;
@@ -191,7 +192,10 @@ public LoadTsFileAnalyzer(
     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();
@@ -275,7 +279,7 @@ private boolean doAsyncLoad(final Analysis analysis) {
               databaseLevel,
               isConvertOnTypeMismatch,
               isVerifySchema,
-              isAutoCreateSchemaRequested,
+              isAutoCreateSchemaEnabled,
               tabletConversionThresholdBytes,
               isGeneratedByPipe,
               context.getSession().getUserName());
@@ -416,10 +420,7 @@ private boolean handleSingleMiniFile(final int i) throws FileNotFoundException {
 
       final TSStatus status =
           loadTsFileDataTypeConverter
-              .convertForTreeModel(
-                  LoadTsFileStatement.createUnchecked(tsFiles.get(i).getPath())
-                      .setDeleteAfterLoad(isDeleteAfterLoad)
-                      .setConvertOnTypeMismatch(isConvertOnTypeMismatch))
+              .convertForTreeModel(createTreeConversionStatement(tsFiles.get(i)))
               .orElse(null);
 
       if (status == null || !loadTsFileDataTypeConverter.isSuccessful(status)) {
@@ -591,6 +592,19 @@ private Analysis executeTabletConversionOnException(
     return analysis;
   }
 
+  private LoadTsFileStatement createTreeConversionStatement(final File tsFile)
+      throws FileNotFoundException {
+    final LoadTsFileStatement statement =
+        LoadTsFileStatement.createUnchecked(tsFile.getPath())
+            .setDeleteAfterLoad(isDeleteAfterLoad)
+            .setConvertOnTypeMismatch(isConvertOnTypeMismatch);
+    statement.setAutoCreateSchema(isAutoCreateSchemaRequested);
+    if (isGeneratedByPipe) {
+      statement.markIsGeneratedByPipe();
+    }
+    return statement;
+  }
+
   private boolean setTemporaryUnavailableStatusIfNecessary(
       final Analysis analysis, final Throwable throwable) {
     // Region unavailability can persist beyond recovery. Retain the file for retry; permanent
@@ -649,6 +663,7 @@ boolean isTemporaryUnavailableDueToPipeSchemaNotReady(final Throwable throwable)
 
   private boolean shouldSkipConversion(LoadAnalyzeException e) {
     return e instanceof LoadAnalyzeInvalidPathException
+        || e instanceof LoadAnalyzeMissingDatabaseException
         || (e instanceof LoadAnalyzeTypeMismatchException)
             && !loadTsFileStatement.isConvertOnTypeMismatch();
   }
@@ -961,6 +976,14 @@ private void autoCreateDatabase()
       }
       databasesNeededToBeSet.removeAll(schemaCache.getAlreadySetDatabases());
 
+      if (!databasesNeededToBeSet.isEmpty() && !isAutoCreateSchemaEnabled) {
+        throw new LoadAnalyzeMissingDatabaseException(
+            String.format(
+                DataNodeQueryMessages
+                    .EXCEPTION_THE_DATABASE_ARG_DOES_NOT_EXIST_AND_AUTO_CREATE_SCHEMA_IS_DISABLED_28F8D49A,
+                databasesNeededToBeSet.iterator().next()));
+      }
+
       // 2. create the databases that do not exist
       for (final PartialPath databasePath : databasesNeededToBeSet) {
         final DatabaseSchemaStatement statement =
@@ -1084,7 +1107,7 @@ private ISchemaTree autoCreateSchema() throws IllegalPathException {
           encodingsList,
           compressionTypesList,
           isAlignedList,
-          isAutoCreateSchemaRequested,
+          isAutoCreateSchemaEnabled,
           context);
     }
 

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

@@ -35,6 +35,8 @@
 import java.util.List;
 import java.util.stream.Collectors;
 
+import static org.apache.iotdb.db.queryengine.plan.analyze.schema.SchemaValidator.isAutoCreateSchemaAllowed;
+
 class NormalSchemaFetcher {
 
   private final IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
@@ -168,7 +170,7 @@ List<Integer> processNormalTimeSeries(
     }
 
     // [Step 5] Auto Create and process the missing schema
-    if (config.isAutoCreateSchemaEnabled()) {
+    if (config.isAutoCreateSchemaEnabled() && isAutoCreateSchemaAllowed(context)) {
       // Check the isAligned value. If the input value is different from the actual value of the
       // existing device, throw exception.
       PartialPath devicePath = schemaComputationWithAutoCreation.getDevicePath();
@@ -319,7 +321,7 @@ void processNormalTimeSeries(
     }
 
     // [Step 5] Auto Create and process the missing schema
-    if (config.isAutoCreateSchemaEnabled()) {
+    if (config.isAutoCreateSchemaEnabled() && isAutoCreateSchemaAllowed(context)) {
       List<PartialPath> devicePathList =
           schemaComputationWithAutoCreationList.stream()
               .map(ISchemaComputationWithAutoCreation::getDevicePath)

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

@@ -37,6 +37,12 @@
 
 public class SchemaValidator {
 
+  static boolean isAutoCreateSchemaAllowed(final MPPQueryContext context) {
+    return context == null
+        || context.getSession() == null
+        || context.getSession().isEnableAutoCreateSchema();
+  }
+
   public static void validate(
       ISchemaFetcher schemaFetcher, InsertBaseStatement insertStatement, MPPQueryContext context) {
     try {
@@ -86,7 +92,7 @@ public static ISchemaTree validate(
       final List<Boolean> isAlignedList,
       final boolean autoCreateSchema,
       final MPPQueryContext context) {
-    return autoCreateSchema
+    return autoCreateSchema && isAutoCreateSchemaAllowed(context)
         ? schemaFetcher.fetchSchemaListWithAutoCreate(
             devicePaths,
             measurements,