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 index 6640277eddbab..d3778b5d528e8 100644 --- 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 @@ -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 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 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 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(); diff --git a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java index c0b6c32d3fce5..c6a839aff0fee 100644 --- a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java +++ b/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 = diff --git a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java index 522bf93d99ad3..f78ee27beaa23 100644 --- a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java +++ b/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 = diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/exception/load/LoadAnalyzeMissingDatabaseException.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/exception/load/LoadAnalyzeMissingDatabaseException.java new file mode 100644 index 0000000000000..2cccebbb9ec3b --- /dev/null +++ b/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); + } +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileAnalyzer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileAnalyzer.java index 569883ea76e3f..bdabeffe235f1 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileAnalyzer.java +++ b/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 @@ 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(); @@ -180,7 +185,10 @@ public LoadTsFileAnalyzer( 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(); @@ -318,13 +326,14 @@ private boolean doAsyncLoad(final IAnalysis analysis) { } else { databaseName = null; } + // The active loader uses a new session, so carry the effective policy with it. final Map activeLoadAttributes = ActiveLoadPathHelper.buildAttributes( databaseName, databaseLevel, isConvertOnTypeMismatch, isVerifySchema, - isAutoCreateSchemaRequested, + isAutoCreateSchemaEnabled, tabletConversionThresholdBytes, isGeneratedByPipe, Objects.nonNull(context) && Objects.nonNull(context.getSession()) @@ -480,8 +489,10 @@ && handleSingleMiniFile(i)) { 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 @@ && handleSingleMiniFile(i)) { 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 @@ private boolean handleSingleMiniFile(final int i) throws FileNotFoundException { 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 TreeSchemaAutoCreatorAndVerifier getOrCreateTreeSchemaVerifier() { 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 @@ private void executeTabletConversionOnException( 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 @@ private void executeTabletConversionOnException( 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 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 @@ boolean isTemporaryUnavailableDueToPipeSchemaNotReady(final Throwable throwable) private boolean shouldSkipConversion(LoadAnalyzeException e) { return e instanceof LoadAnalyzeInvalidPathException + || e instanceof LoadAnalyzeMissingDatabaseException || (e instanceof LoadAnalyzeTypeMismatchException) && !isConvertOnTypeMismatch; } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileTableSchemaCache.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileTableSchemaCache.java index e4078a8dc190e..7f3d11c4f71b0 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileTableSchemaCache.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileTableSchemaCache.java @@ -33,6 +33,7 @@ import org.apache.iotdb.db.conf.IoTDBConfig; import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.exception.load.LoadAnalyzeException; +import org.apache.iotdb.db.exception.load.LoadAnalyzeMissingDatabaseException; import org.apache.iotdb.db.exception.load.LoadAnalyzeTypeMismatchException; import org.apache.iotdb.db.exception.load.LoadRuntimeOutOfMemoryException; import org.apache.iotdb.db.i18n.DataNodeQueryMessages; @@ -106,6 +107,7 @@ public class LoadTsFileTableSchemaCache { private final Metadata metadata; private final MPPQueryContext context; private final boolean shouldVerifyDataType; + private final boolean allowAutoCreateSchema; private Map> currentBatchTable2Devices; @@ -129,12 +131,28 @@ public LoadTsFileTableSchemaCache( final boolean needToCreateDatabase, final boolean shouldVerifyDataType) throws LoadRuntimeOutOfMemoryException { + this(metadata, context, needToCreateDatabase, shouldVerifyDataType, true); + } + + public LoadTsFileTableSchemaCache( + final Metadata metadata, + final MPPQueryContext context, + final boolean needToCreateDatabase, + final boolean shouldVerifyDataType, + final boolean allowAutoCreateSchema) + throws LoadRuntimeOutOfMemoryException { this.block = LoadTsFileMemoryManager.getInstance() .allocateMemoryBlock(ANALYZE_SCHEMA_MEMORY_SIZE_IN_BYTES); this.metadata = metadata; this.context = context; this.shouldVerifyDataType = shouldVerifyDataType; + this.allowAutoCreateSchema = + allowAutoCreateSchema + && IoTDBDescriptor.getInstance().getConfig().isAutoCreateSchemaEnabled() + && (context == null + || context.getSession() == null + || context.getSession().isEnableAutoCreateSchema()); this.currentBatchTable2Devices = new HashMap<>(); this.currentModifications = PatternTreeMapFactory.getModsPatternTreeMap(); this.needToCreateDatabase = needToCreateDatabase; @@ -358,7 +376,12 @@ public void createTableAndDatabaseIfNecessary(final String tableName) final TableSchema realSchema = metadata .validateTableHeaderSchema4TsFile( - database, fileSchema, context, true, true, needDecode4DifferentTimeColumn) + database, + fileSchema, + context, + allowAutoCreateSchema, + true, + needDecode4DifferentTimeColumn) .orElse(null); if (Objects.isNull(realSchema)) { throw new LoadAnalyzeException( @@ -394,11 +417,11 @@ private void autoCreateTableDatabaseIfAbsent(final String database) throws LoadA return; } - if (!IoTDBDescriptor.getInstance().getConfig().isAutoCreateSchemaEnabled()) { - throw new LoadAnalyzeException( + if (!allowAutoCreateSchema) { + throw new LoadAnalyzeMissingDatabaseException( String.format( DataNodeQueryMessages - .QUERY_EXCEPTION_THE_DATABASE_S_DOES_NOT_EXIST_PLEASE_ENABLE_ENABLE_AUTO_B6683D0E, + .EXCEPTION_THE_DATABASE_ARG_DOES_NOT_EXIST_AND_AUTO_CREATE_SCHEMA_IS_DISABLED_28F8D49A, database)); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/TreeSchemaAutoCreatorAndVerifier.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/TreeSchemaAutoCreatorAndVerifier.java index 555756d64526f..d27da400dfa13 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/TreeSchemaAutoCreatorAndVerifier.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/TreeSchemaAutoCreatorAndVerifier.java @@ -39,6 +39,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.LoadFileException; @@ -421,6 +422,14 @@ private void autoCreateDatabase() } databasesNeededToBeSet.removeAll(schemaCache.getAlreadySetDatabases()); + if (!databasesNeededToBeSet.isEmpty() && !loadTsFileAnalyzer.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 = @@ -540,7 +549,7 @@ private ISchemaTree autoCreateSchema() throws IllegalPathException { encodingsList, compressionTypesList, isAlignedList, - loadTsFileAnalyzer.isAutoCreateSchemaRequested(), + loadTsFileAnalyzer.isAutoCreateSchemaEnabled(), loadTsFileAnalyzer.context); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/schema/NormalSchemaFetcher.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/schema/NormalSchemaFetcher.java index 04206a7f2083f..3beede1a6b246 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/schema/NormalSchemaFetcher.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/schema/NormalSchemaFetcher.java @@ -37,6 +37,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(); @@ -170,13 +172,14 @@ List processNormalTimeSeries( } // [Step 5] Auto Create and process the missing schema - if (config.isAutoCreateSchemaEnabled() - || schemaComputationWithAutoCreation - .getDevicePath() - .startsWith("root." + SystemConstant.SYSTEM_PREFIX_KEY) - || schemaComputationWithAutoCreation - .getDevicePath() - .startsWith("root." + SystemConstant.AUDIT_PREFIX_KEY)) { + if (isAutoCreateSchemaAllowed(context) + && (config.isAutoCreateSchemaEnabled() + || schemaComputationWithAutoCreation + .getDevicePath() + .startsWith("root." + SystemConstant.SYSTEM_PREFIX_KEY) + || schemaComputationWithAutoCreation + .getDevicePath() + .startsWith("root." + SystemConstant.AUDIT_PREFIX_KEY))) { // Check the isAligned value. If the input value is different from the actual value of the // existing device, throw exception. PartialPath devicePath = schemaComputationWithAutoCreation.getDevicePath(); @@ -323,6 +326,17 @@ void processNormalTimeSeries( return; } + if (!isAutoCreateSchemaAllowed(context)) { + for (int i = 0; i < indexOfDevicesNeedAutoCreateSchema.size(); i++) { + final ISchemaComputationWithAutoCreation computation = + schemaComputationWithAutoCreationList.get(indexOfDevicesNeedAutoCreateSchema.get(i)); + for (final int index : indexOfMeasurementsNeedAutoCreate.get(i)) { + computation.computeMeasurement(index, null); + } + } + return; + } + List copyOfIndexOfDevicesNeedAutoCreateSchema = new ArrayList<>(indexOfDevicesNeedAutoCreateSchema); if (!config.isAutoCreateSchemaEnabled()) { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/schema/SchemaValidator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/schema/SchemaValidator.java index 02817fb424528..d2ce31d184aff 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/schema/SchemaValidator.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/schema/SchemaValidator.java @@ -56,6 +56,12 @@ public class SchemaValidator { private static final Logger LOGGER = LoggerFactory.getLogger(SchemaValidator.class); + 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 { @@ -189,7 +195,7 @@ public static ISchemaTree validate( final List isAlignedList, final boolean autoCreateSchema, final MPPQueryContext context) { - return autoCreateSchema + return autoCreateSchema && isAutoCreateSchemaAllowed(context) ? schemaFetcher.fetchSchemaListWithAutoCreate( devicePaths, measurements, diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/schema/TemplateSchemaFetcher.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/schema/TemplateSchemaFetcher.java index bca0b54ff1e6c..4b94613b2f1f3 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/schema/TemplateSchemaFetcher.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/schema/TemplateSchemaFetcher.java @@ -38,6 +38,7 @@ import java.util.Map; import java.util.stream.Collectors; +import static org.apache.iotdb.db.queryengine.plan.analyze.schema.SchemaValidator.isAutoCreateSchemaAllowed; import static org.apache.iotdb.db.utils.EncodingInferenceUtils.getDefaultEncoding; class TemplateSchemaFetcher { @@ -73,7 +74,9 @@ List processTemplateTimeSeries( } } - if (!extensionMeasurementList.isEmpty() && config.isAutoCreateSchemaEnabled()) { + if (!extensionMeasurementList.isEmpty() + && config.isAutoCreateSchemaEnabled() + && isAutoCreateSchemaAllowed(context)) { autoCreateSchemaExecutor.autoExtendTemplate( template.getName(), extensionMeasurementList, extensionDataTypeList, context); } @@ -110,7 +113,7 @@ List processTemplateTimeSeries( // not activated // auto create and process the missing schema - if (config.isAutoCreateSchemaEnabled()) { + if (config.isAutoCreateSchemaEnabled() && isAutoCreateSchemaAllowed(context)) { ClusterSchemaTree schemaTree = new ClusterSchemaTree(); autoCreateSchemaExecutor.autoActivateTemplate( schemaTree, devicePath, template.getId(), context); @@ -159,7 +162,9 @@ void processTemplateTimeSeries( } } } - if (!extensionMeasurementMap.isEmpty() && config.isAutoCreateSchemaEnabled()) { + if (!extensionMeasurementMap.isEmpty() + && config.isAutoCreateSchemaEnabled() + && isAutoCreateSchemaAllowed(context)) { autoCreateSchemaExecutor.autoExtendTemplate(extensionMeasurementMap, context); } @@ -211,7 +216,7 @@ void processTemplateTimeSeries( } // auto create and process the missing schema - if (config.isAutoCreateSchemaEnabled()) { + if (config.isAutoCreateSchemaEnabled() && isAutoCreateSchemaAllowed(context)) { ClusterSchemaTree schemaTree = new ClusterSchemaTree(); autoCreateSchemaExecutor.autoActivateTemplate( schemaTree, diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/load/LoadSingleTsFileNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/load/LoadSingleTsFileNode.java index ecaf0aea6ec2f..4bdcdc15674b8 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/load/LoadSingleTsFileNode.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/load/LoadSingleTsFileNode.java @@ -62,6 +62,7 @@ public class LoadSingleTsFileNode extends WritePlanNode { private final boolean isTableModel; private final String database; private final boolean deleteAfterLoad; + private final boolean autoCreateSchema; private final long writePointCount; private boolean needDecodeTsFile; @@ -75,12 +76,33 @@ public LoadSingleTsFileNode( final boolean deleteAfterLoad, final long writePointCount, final boolean needDecodeTsFile) { + this( + id, + resource, + isTableModel, + database, + deleteAfterLoad, + writePointCount, + needDecodeTsFile, + true); + } + + public LoadSingleTsFileNode( + final PlanNodeId id, + final TsFileResource resource, + final boolean isTableModel, + final String database, + final boolean deleteAfterLoad, + final long writePointCount, + final boolean needDecodeTsFile, + final boolean autoCreateSchema) { super(id); this.tsFile = resource.getTsFile(); this.resource = resource; this.isTableModel = isTableModel; this.database = database; this.deleteAfterLoad = deleteAfterLoad; + this.autoCreateSchema = autoCreateSchema; this.writePointCount = writePointCount; this.needDecodeTsFile = needDecodeTsFile; } @@ -163,6 +185,10 @@ public boolean isTableModel() { return isTableModel; } + public boolean isAutoCreateSchema() { + return autoCreateSchema; + } + public long getWritePointCount() { return writePointCount; } @@ -278,6 +304,7 @@ public boolean equals(Object o) { && Objects.equals(database, loadSingleTsFileNode.database) && Objects.equals(needDecodeTsFile, loadSingleTsFileNode.needDecodeTsFile) && Objects.equals(deleteAfterLoad, loadSingleTsFileNode.deleteAfterLoad) + && autoCreateSchema == loadSingleTsFileNode.autoCreateSchema && Objects.equals(localRegionReplicaSet, loadSingleTsFileNode.localRegionReplicaSet); } @@ -290,6 +317,7 @@ public int hashCode() { database, needDecodeTsFile, deleteAfterLoad, + autoCreateSchema, localRegionReplicaSet); } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/load/LoadTsFileNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/load/LoadTsFileNode.java index 6b1a6b464f146..929f4a1fed121 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/load/LoadTsFileNode.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/load/LoadTsFileNode.java @@ -130,7 +130,8 @@ private List splitByPartitionForTreeModel(Analysis analysis) { database, statement.isDeleteAfterLoad(), statement.getWritePointCount(i), - needDecode4TimeColumn)); + needDecode4TimeColumn, + statement.isAutoCreateSchema())); } return res; } @@ -153,7 +154,8 @@ private List splitByPartitionForTableModel( database, statement.isDeleteAfterLoad(), statement.getWritePointCount(i), - needDecode4TimeColumn)); + needDecode4TimeColumn, + statement.isAutoCreateSchema())); } else { throw new IllegalStateException( DataNodeQueryMessages.LOADTSFILE_STATEMENT_IS_NULL_DURING_TABLE_MODEL_SPLIT); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/Metadata.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/Metadata.java index 04db571083f51..58a4c146b17a1 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/Metadata.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/Metadata.java @@ -116,6 +116,7 @@ DeviceEntryDataSetResult indexScan( *

The caller need to recheck the dataType of measurement columns to decide whether to do * partial insert * + * @param allowAutoCreateSchema whether to allow automatic creation of missing tables and columns * @param isStrictTagColumn if {@code true}, when the table already exists, the tag columns in the * existing table should be the prefix of those in the input tableSchema, or input tag columns * be the prefix of existing tag columns. @@ -129,7 +130,7 @@ Optional validateTableHeaderSchema4TsFile( final String database, final TableSchema tableSchema, final MPPQueryContext context, - final boolean allowCreateTable, + final boolean allowAutoCreateSchema, final boolean isStrictTagColumn, final AtomicBoolean needDecode4DifferentTimeColumn) throws LoadAnalyzeTableColumnDisorderException; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/TableMetadataImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/TableMetadataImpl.java index 04dcc93b7312c..3fc7d57bdab24 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/TableMetadataImpl.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/TableMetadataImpl.java @@ -1705,7 +1705,7 @@ public Optional validateTableHeaderSchema4TsFile( final String database, final TableSchema tableSchema, final MPPQueryContext context, - final boolean allowCreateTable, + final boolean allowAutoCreateSchema, final boolean isStrictTagColumn, final AtomicBoolean needDecode4DifferentTimeColumn) throws LoadAnalyzeTableColumnDisorderException { @@ -1714,7 +1714,7 @@ public Optional validateTableHeaderSchema4TsFile( database, tableSchema, context, - allowCreateTable, + allowAutoCreateSchema, isStrictTagColumn, needDecode4DifferentTimeColumn); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/TableHeaderSchemaValidator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/TableHeaderSchemaValidator.java index 01c2555608ad9..797085491206c 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/TableHeaderSchemaValidator.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/TableHeaderSchemaValidator.java @@ -103,7 +103,7 @@ public Optional validateTableHeaderSchema4TsFile( final String database, final TableSchema tableSchema, final MPPQueryContext context, - final boolean allowCreateTable, + final boolean allowAutoCreateSchema, final boolean isStrictTagColumn, final @Nonnull AtomicBoolean needDecode4DifferentTimeColumn) throws LoadAnalyzeTableColumnDisorderException { @@ -124,13 +124,15 @@ public Optional validateTableHeaderSchema4TsFile( final List missingColumnList = new ArrayList<>(); final boolean isAutoCreateSchemaEnabled = - IoTDBDescriptor.getInstance().getConfig().isAutoCreateSchemaEnabled(); + allowAutoCreateSchema + && IoTDBDescriptor.getInstance().getConfig().isAutoCreateSchemaEnabled() + && context.getSession().isEnableAutoCreateSchema(); // first round validate, check existing schema if (table == null) { // auto create missing table // it's ok that many write requests concurrently auto create same table, the thread safety // will be guaranteed by ProcedureManager.createTable in CN - if (allowCreateTable && isAutoCreateSchemaEnabled) { + if (isAutoCreateSchemaEnabled) { autoCreateTable(context, database, tableSchema); table = DataNodeTableCache.getInstance().getTable(database, tableSchema.getTableName(), false); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadFallbackHandler.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadFallbackHandler.java index 4d0752d3f8639..5204941451738 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadFallbackHandler.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadFallbackHandler.java @@ -38,9 +38,12 @@ import java.util.Collections; import java.util.Iterator; import java.util.List; +import java.util.Map; import java.util.Objects; import java.util.stream.Collectors; +import static org.apache.iotdb.db.storageengine.load.config.LoadTsFileConfigurator.AUTO_CREATE_SCHEMA_KEY; + /** * Fallback execution handler for the LOAD scheduler. When direct chunk staging fails, converts * failed TsFiles into In-Memory Tablets, retries insertion through the regular write pipeline, and @@ -190,10 +193,12 @@ private static TSStatus executeTableModelConversion( final String filePath, final boolean isGeneratedByPipe, final boolean deleteAfterLoad) { + final Map attributes = + Collections.singletonMap(AUTO_CREATE_SCHEMA_KEY, String.valueOf(node.isAutoCreateSchema())); final LoadTsFile statement = (isGeneratedByPipe - ? LoadTsFile.createForPipe(null, filePath, Collections.emptyMap()) - : LoadTsFile.createUnchecked(null, filePath, Collections.emptyMap())) + ? LoadTsFile.createForPipe(null, filePath, attributes) + : LoadTsFile.createUnchecked(null, filePath, attributes)) .setDatabase(node.getDatabase()) .setDeleteAfterLoad(deleteAfterLoad) .setConvertOnTypeMismatch(true); @@ -214,6 +219,7 @@ private static TSStatus executeTreeModelConversion( final String database = LoadTsFileScheduler.getPartitionQueryDatabase(node, isGeneratedByPipe); final LoadTsFileStatement statement = buildRetryTreeLoadStatement(filePath, deleteAfterLoad, database, isGeneratedByPipe); + statement.setAutoCreateSchema(node.isAutoCreateSchema()); return converter.convertForTreeModel(statement).orElse(null); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/converter/LoadTsFileDataTypeConverter.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/converter/LoadTsFileDataTypeConverter.java index 20d2802e0fdba..f78d6c5c53715 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/converter/LoadTsFileDataTypeConverter.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/converter/LoadTsFileDataTypeConverter.java @@ -24,6 +24,7 @@ import org.apache.iotdb.commons.conf.IoTDBConstant; import org.apache.iotdb.commons.exception.pipe.PipeRuntimeOutOfMemoryCriticalException; import org.apache.iotdb.commons.pipe.config.PipeConfig; +import org.apache.iotdb.commons.queryengine.common.SessionInfo; import org.apache.iotdb.commons.queryengine.common.SqlDialect; import org.apache.iotdb.db.auth.AuthorityChecker; import org.apache.iotdb.db.conf.IoTDBDescriptor; @@ -157,20 +158,11 @@ private static class TabletConversionSemaphoreHolder { private final MPPQueryContext context; private final SqlParser relationalSqlParser = new SqlParser(); - private final LoadTableStatementDataTypeConvertExecutionVisitor - tableStatementDataTypeConvertExecutionVisitor; - private final LoadTreeStatementDataTypeConvertExecutionVisitor - treeStatementDataTypeConvertExecutionVisitor; public LoadTsFileDataTypeConverter( final MPPQueryContext context, final boolean isGeneratedByPipe) { this.context = context; this.isGeneratedByPipe = isGeneratedByPipe; - - tableStatementDataTypeConvertExecutionVisitor = - new LoadTableStatementDataTypeConvertExecutionVisitor(this::executeForTableModel); - treeStatementDataTypeConvertExecutionVisitor = - new LoadTreeStatementDataTypeConvertExecutionVisitor(this::executeForTreeModel); } public Optional convertForTableModel(final LoadTsFile loadTsFileTableStatement) { @@ -179,7 +171,11 @@ public Optional convertForTableModel(final LoadTsFile loadTsFileTableS getTabletConversionSemaphore().acquire(); isPermitAcquired = true; return loadTsFileTableStatement.accept( - tableStatementDataTypeConvertExecutionVisitor, loadTsFileTableStatement.getDatabase()); + new LoadTableStatementDataTypeConvertExecutionVisitor( + (statement, databaseName) -> + executeForTableModel( + statement, databaseName, loadTsFileTableStatement.isAutoCreateSchema())), + loadTsFileTableStatement.getDatabase()); } catch (final InterruptedException e) { return getInterruptedConversionStatus(e); } catch (Exception e) { @@ -240,12 +236,16 @@ private IClientSession createConversionSession(final SqlDialect sqlDialect) { session.setClientVersion(IoTDBConstant.ClientVersion.V_1_0); session.setZoneId(ZoneId.systemDefault()); session.setSqlDialect(sqlDialect); + if (Objects.nonNull(context) && Objects.nonNull(context.getSession())) { + session.setEnableAutoCreateSchema(context.getSession().isEnableAutoCreateSchema()); + } SESSION_MANAGER.registerSession(session); return session; } - private TSStatus executeForTableModel(final Statement statement, final String databaseName) { + private TSStatus executeForTableModel( + final Statement statement, final String databaseName, final boolean allowAutoCreateSchema) { final IClientSession currentSession = SESSION_MANAGER.getCurrSession(); final boolean needToCreateSession = Objects.isNull(currentSession); final IClientSession session = @@ -257,7 +257,15 @@ private TSStatus executeForTableModel(final Statement statement, final String da relationalSqlParser, session, SESSION_MANAGER.requestQueryId(), - SESSION_MANAGER.getSessionInfoOfPipeReceiver(session, databaseName), + new SessionInfo( + session.getId(), + new UserEntity( + session.getUserId(), session.getUsername(), session.getClientAddress()), + ZoneId.systemDefault(), + session.getClientVersion(), + databaseName, + SqlDialect.TABLE, + allowAutoCreateSchema && session.isEnableAutoCreateSchema()), "", LocalExecutionPlanner.getInstance().metadata, IoTDBDescriptor.getInstance().getConfig().getQueryTimeoutThreshold()) @@ -275,7 +283,11 @@ public Optional convertForTreeModel(final LoadTsFileStatement loadTsFi try { getTabletConversionSemaphore().acquire(); isPermitAcquired = true; - return loadTsFileTreeStatement.accept(treeStatementDataTypeConvertExecutionVisitor, null); + return loadTsFileTreeStatement.accept( + new LoadTreeStatementDataTypeConvertExecutionVisitor( + statement -> + executeForTreeModel(statement, loadTsFileTreeStatement.isAutoCreateSchema())), + null); } catch (final InterruptedException e) { return getInterruptedConversionStatus(e); } catch (Exception e) { @@ -298,7 +310,8 @@ public Optional convertForTreeModel(final LoadTsFileStatement loadTsFi } } - private TSStatus executeForTreeModel(final Statement statement) { + private TSStatus executeForTreeModel( + final Statement statement, final boolean allowAutoCreateSchema) { final IClientSession currentSession = SESSION_MANAGER.getCurrSession(); final boolean needToCreateSession = Objects.isNull(currentSession); final IClientSession session = @@ -309,7 +322,19 @@ private TSStatus executeForTreeModel(final Statement statement) { .executeForTreeModel( isGeneratedByPipe ? new PipeEnrichedStatement(statement) : statement, SESSION_MANAGER.requestQueryId(), - SESSION_MANAGER.getSessionInfoOfTreeModel(session), + new SessionInfo( + session.getId(), + new UserEntity( + session.getUserId(), session.getUsername(), session.getClientAddress()), + ZoneId.systemDefault(), + session.getClientVersion(), + session.getDatabaseName(), + SqlDialect.TREE, + allowAutoCreateSchema + && session.isEnableAutoCreateSchema() + && (Objects.isNull(context) + || Objects.isNull(context.getSession()) + || context.getSession().isEnableAutoCreateSchema())), "", ClusterPartitionFetcher.getInstance(), ClusterSchemaFetcher.getInstance(), diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileAnalyzerDiagnosticsTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileAnalyzerDiagnosticsTest.java index 3dade21bb9926..5f00297eb5cd1 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileAnalyzerDiagnosticsTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileAnalyzerDiagnosticsTest.java @@ -21,21 +21,40 @@ import org.apache.iotdb.common.rpc.thrift.TEndPoint; import org.apache.iotdb.common.rpc.thrift.TSStatus; +import org.apache.iotdb.commons.audit.UserEntity; +import org.apache.iotdb.commons.client.IClientManager; +import org.apache.iotdb.commons.conf.IoTDBConstant; +import org.apache.iotdb.commons.consensus.ConfigRegionId; import org.apache.iotdb.commons.exception.IoTDBException; import org.apache.iotdb.commons.exception.IoTDBRuntimeException; import org.apache.iotdb.commons.exception.SemanticException; +import org.apache.iotdb.commons.path.PartialPath; +import org.apache.iotdb.commons.queryengine.common.SessionInfo; +import org.apache.iotdb.commons.queryengine.common.SqlDialect; +import org.apache.iotdb.confignode.rpc.thrift.TShowDatabaseResp; import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.exception.load.LoadAnalyzeException; +import org.apache.iotdb.db.exception.load.LoadAnalyzeMissingDatabaseException; +import org.apache.iotdb.db.exception.load.LoadAnalyzeMissingSchemaException; import org.apache.iotdb.db.i18n.DataNodeQueryMessages; +import org.apache.iotdb.db.protocol.client.ConfigNodeClient; import org.apache.iotdb.db.queryengine.common.MPPQueryContext; import org.apache.iotdb.db.queryengine.common.QueryId; +import org.apache.iotdb.db.queryengine.plan.Coordinator; import org.apache.iotdb.db.queryengine.plan.analyze.Analysis; import org.apache.iotdb.db.queryengine.plan.analyze.IAnalysis; import org.apache.iotdb.db.queryengine.plan.analyze.schema.SchemaValidator; +import org.apache.iotdb.db.queryengine.plan.execution.config.executor.ClusterConfigTaskExecutor; +import org.apache.iotdb.db.queryengine.plan.relational.metadata.Metadata; +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.schemaengine.table.DataNodeTableCache; +import org.apache.iotdb.db.storageengine.load.active.ActiveLoadPathHelper; import org.apache.iotdb.db.storageengine.load.converter.LoadTsFileDataTypeConverter; +import org.apache.iotdb.db.storageengine.load.util.LoadUtil; import org.apache.iotdb.rpc.TSStatusCode; +import org.apache.tsfile.enums.ColumnCategory; import org.apache.tsfile.enums.TSDataType; import org.apache.tsfile.file.metadata.IDeviceID; import org.apache.tsfile.file.metadata.TableSchema; @@ -48,6 +67,7 @@ import org.junit.Test; import org.junit.rules.TemporaryFolder; import org.junit.runner.RunWith; +import org.mockito.ArgumentCaptor; import org.powermock.api.mockito.PowerMockito; import org.powermock.core.classloader.annotations.PowerMockIgnore; import org.powermock.core.classloader.annotations.PrepareForTest; @@ -57,18 +77,38 @@ import java.io.File; import java.io.FileOutputStream; import java.io.IOException; +import java.time.ZoneOffset; import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; import java.util.Optional; +import java.util.concurrent.atomic.AtomicReference; +import static org.apache.iotdb.db.storageengine.load.config.LoadTsFileConfigurator.ASYNC_LOAD_KEY; +import static org.apache.iotdb.db.storageengine.load.config.LoadTsFileConfigurator.AUTO_CREATE_SCHEMA_KEY; +import static org.apache.iotdb.db.storageengine.load.config.LoadTsFileConfigurator.DATABASE_NAME_KEY; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyBoolean; import static org.mockito.ArgumentMatchers.anyList; +import static org.mockito.ArgumentMatchers.anyMap; +import static org.mockito.Mockito.clearInvocations; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyZeroInteractions; import static org.mockito.Mockito.when; @RunWith(PowerMockRunner.class) @PowerMockIgnore({"com.sun.org.apache.xerces.*", "javax.xml.*", "org.xml.*", "javax.management.*"}) -@PrepareForTest({LoadTsFileAnalyzer.class, SchemaValidator.class}) +@PrepareForTest({ + LoadTsFileAnalyzer.class, + LoadUtil.class, + SchemaValidator.class, + DataNodeTableCache.class, + Coordinator.class, + ClusterConfigTaskExecutor.class +}) public class LoadTsFileAnalyzerDiagnosticsTest { @Rule public TemporaryFolder temporaryFolder = new TemporaryFolder(); @@ -301,6 +341,381 @@ public void testConversionStatusKeepsCodeAndMetadataWithoutMutatingOriginal() th Assert.assertTrue(status.getMessage().contains("missing WRITE_DATA privilege")); } + @Test + public void testMiniFileTableConversionPreservesAutoCreateSchema() throws Exception { + assertTableConversionPreservesAutoCreateSchema(true); + } + + @Test + public void testFallbackTableConversionPreservesAutoCreateSchema() throws Exception { + assertTableConversionPreservesAutoCreateSchema(false); + } + + @Test + public void testMiniFileTreeConversionPreservesAutoCreateSchema() throws Exception { + assertTreeConversionPreservesAutoCreateSchema(true); + } + + @Test + public void testFallbackTreeConversionPreservesAutoCreateSchema() throws Exception { + assertTreeConversionPreservesAutoCreateSchema(false); + } + + private void assertTreeConversionPreservesAutoCreateSchema(final boolean miniFile) + throws Exception { + when(converter.convertForTreeModel(any())) + .thenReturn(Optional.of(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()))); + final File file = temporaryFolder.newFile("tree-conversion.tsfile"); + for (final boolean tableModelRequest : new boolean[] {false, true}) { + for (final boolean requested : new boolean[] {false, true, false}) { + for (final boolean generatedByPipe : new boolean[] {false, true}) { + clearInvocations(converter); + final Map attributes = + Collections.singletonMap(AUTO_CREATE_SCHEMA_KEY, String.valueOf(requested)); + final LoadTsFile tableStatement = + LoadTsFile.createUnchecked(null, file.getAbsolutePath(), attributes); + final LoadTsFileStatement treeStatement = + LoadTsFileStatement.createUnchecked(file.getAbsolutePath()); + treeStatement.setAutoCreateSchema(requested); + final MPPQueryContext context = new MPPQueryContext(new QueryId("tree_conversion")); + try (final LoadTsFileAnalyzer analyzer = + tableModelRequest + ? new LoadTsFileAnalyzer(tableStatement, generatedByPipe, context) + : new LoadTsFileAnalyzer(treeStatement, generatedByPipe, context)) { + final List tableModelFiles = + Whitebox.getInternalState(analyzer, "isTableModelTsFile"); + tableModelFiles.set(0, false); + Whitebox.setInternalState(analyzer, "isTableModelTsFileReliableIndex", 0); + if (miniFile) { + Assert.assertTrue( + (Boolean) Whitebox.invokeMethod(analyzer, "handleSingleMiniFile", 0)); + } else { + final Analysis analysis = new Analysis(); + Whitebox.invokeMethod( + analyzer, + "executeTabletConversionOnException", + (IAnalysis) analysis, + new LoadAnalyzeException("original analysis failure")); + Assert.assertTrue(analysis.isFinishQueryAfterAnalyze()); + Assert.assertFalse(analysis.isFailed()); + } + final ArgumentCaptor statement = + ArgumentCaptor.forClass(LoadTsFileStatement.class); + verify(converter).convertForTreeModel(statement.capture()); + Assert.assertEquals(requested, statement.getValue().isAutoCreateSchema()); + Assert.assertEquals(Collections.singletonList(file), statement.getValue().getTsFiles()); + } + } + } + } + } + + @Test + public void testMissingDatabaseDoesNotTryTabletConversion() throws Exception { + final File file = temporaryFolder.newFile("missing-database-conversion.tsfile"); + final LoadTsFileStatement statement = LoadTsFileStatement.createUnchecked(file.getPath()); + statement.setAutoCreateSchema(false); + final String message = + String.format( + DataNodeQueryMessages + .EXCEPTION_THE_DATABASE_ARG_DOES_NOT_EXIST_AND_AUTO_CREATE_SCHEMA_IS_DISABLED_28F8D49A, + "missing_db"); + try (final LoadTsFileAnalyzer analyzer = + new LoadTsFileAnalyzer( + statement, false, new MPPQueryContext(new QueryId("missing_database_conversion")))) { + final Analysis analysis = new Analysis(); + Whitebox.invokeMethod( + analyzer, + "executeTabletConversionOnException", + (IAnalysis) analysis, + new LoadAnalyzeMissingDatabaseException(message)); + Assert.assertTrue(analysis.isFinishQueryAfterAnalyze()); + Assert.assertEquals( + TSStatusCode.LOAD_FILE_ERROR.getStatusCode(), analysis.getFailStatus().getCode()); + Assert.assertEquals(message, analysis.getFailStatus().getMessage()); + verifyZeroInteractions(converter); + Assert.assertTrue(file.exists()); + } + } + + @Test + public void testDatabaseAutoCreationHonorsRequestPolicy() throws Exception { + assertDatabaseAutoCreationDisabled(true, false, true); + } + + @Test + public void testDatabaseAutoCreationHonorsSessionPolicy() throws Exception { + assertDatabaseAutoCreationDisabled(true, true, false); + } + + @Test + public void testDatabaseAutoCreationHonorsGlobalPolicy() throws Exception { + assertDatabaseAutoCreationDisabled(false, true, true); + } + + private void assertDatabaseAutoCreationDisabled( + final boolean globalEnabled, final boolean requested, final boolean sessionEnabled) + throws Exception { + final boolean originalGlobalEnabled = + IoTDBDescriptor.getInstance().getConfig().isAutoCreateSchemaEnabled(); + final IClientManager originalClientManager = + Whitebox.getInternalState( + TreeSchemaAutoCreatorAndVerifier.class, "CONFIG_NODE_CLIENT_MANAGER"); + final IClientManager clientManager = + mock(IClientManager.class); + final ConfigNodeClient client = mock(ConfigNodeClient.class); + when(clientManager.borrowClient(any())).thenReturn(client); + when(client.showDatabase(any())) + .thenReturn(new TShowDatabaseResp().setDatabaseInfoMap(Collections.emptyMap())); + final DataNodeTableCache tableCache = mock(DataNodeTableCache.class); + PowerMockito.mockStatic(DataNodeTableCache.class); + when(DataNodeTableCache.getInstance()).thenReturn(tableCache); + final ClusterConfigTaskExecutor executor = mock(ClusterConfigTaskExecutor.class); + PowerMockito.mockStatic(ClusterConfigTaskExecutor.class); + when(ClusterConfigTaskExecutor.getInstance()).thenReturn(executor); + final Coordinator coordinator = mock(Coordinator.class); + PowerMockito.mockStatic(Coordinator.class); + when(Coordinator.getInstance()).thenReturn(coordinator); + final Metadata metadata = mock(Metadata.class); + try { + IoTDBDescriptor.getInstance().getConfig().setAutoCreateSchemaEnabled(globalEnabled); + Whitebox.setInternalState( + TreeSchemaAutoCreatorAndVerifier.class, "CONFIG_NODE_CLIENT_MANAGER", clientManager); + for (final boolean tableModelRequest : new boolean[] {false, true}) { + final File file = + temporaryFolder.newFile("missing-database-" + tableModelRequest + ".tsfile"); + final Map attributes = + Collections.singletonMap(AUTO_CREATE_SCHEMA_KEY, String.valueOf(requested)); + final LoadTsFile tableStatement = + LoadTsFile.createUnchecked(null, file.getAbsolutePath(), attributes); + final LoadTsFileStatement treeStatement = + LoadTsFileStatement.createUnchecked(file.getAbsolutePath()); + treeStatement.setAutoCreateSchema(requested); + // Leave the database-creation path enabled, as it is by default with global creation on. + tableStatement.setAutoCreateDatabase(true); + treeStatement.setAutoCreateDatabase(true); + final MPPQueryContext context = new MPPQueryContext(new QueryId("missing_database")); + context.setSession( + new SessionInfo( + 0, + new UserEntity(0, "root", ""), + ZoneOffset.UTC, + IoTDBConstant.ClientVersion.V_1_0, + "missing_db", + tableModelRequest ? SqlDialect.TABLE : SqlDialect.TREE, + sessionEnabled)); + try (final LoadTsFileAnalyzer analyzer = + tableModelRequest + ? new LoadTsFileAnalyzer(tableStatement, false, context) + : new LoadTsFileAnalyzer(treeStatement, false, context)) { + Assert.assertFalse(analyzer.isAutoCreateSchemaEnabled()); + Whitebox.setInternalState(analyzer, "metadata", metadata); + final LoadTsFileTableSchemaCache tableSchemaCache = + Whitebox.invokeMethod(analyzer, "getOrCreateTableSchemaCache"); + tableSchemaCache.setDatabase("missing_db"); + tableSchemaCache.setTableSchemaMap( + new HashMap<>( + Collections.singletonMap( + "table1", + new TableSchema( + "table1", + Collections.singletonList(new MeasurementSchema("s", TSDataType.INT64)), + Collections.singletonList(ColumnCategory.FIELD))))); + final LoadAnalyzeMissingSchemaException tableFailure = + Assert.assertThrows( + LoadAnalyzeMissingSchemaException.class, + () -> tableSchemaCache.createTableAndDatabaseIfNecessary("table1")); + Assert.assertEquals( + String.format( + DataNodeQueryMessages + .EXCEPTION_THE_DATABASE_ARG_DOES_NOT_EXIST_AND_AUTO_CREATE_SCHEMA_IS_DISABLED_28F8D49A, + "missing_db"), + tableFailure.getMessage()); + final TreeSchemaAutoCreatorAndVerifier verifier = + Whitebox.invokeMethod(analyzer, "getOrCreateTreeSchemaVerifier"); + final LoadTsFileTreeSchemaCache treeCache = + Whitebox.getInternalState(verifier, "schemaCache"); + treeCache.addTimeSeries( + IDeviceID.Factory.DEFAULT_FACTORY.create("root.missing_db.d"), + new MeasurementSchema("s", TSDataType.INT64)); + final LoadAnalyzeMissingSchemaException treeFailure = + Assert.assertThrows( + LoadAnalyzeMissingSchemaException.class, + () -> Whitebox.invokeMethod(verifier, "autoCreateDatabase")); + Assert.assertEquals( + String.format( + DataNodeQueryMessages + .EXCEPTION_THE_DATABASE_ARG_DOES_NOT_EXIST_AND_AUTO_CREATE_SCHEMA_IS_DISABLED_28F8D49A, + "root.missing_db"), + treeFailure.getMessage()); + // An existing database remains usable under the same disabled policy. + treeCache.addAlreadySetDatabase(new PartialPath("root.missing_db")); + Whitebox.invokeMethod(verifier, "autoCreateDatabase"); + Assert.assertEquals(sessionEnabled, context.getSession().isEnableAutoCreateSchema()); + } + } + verifyZeroInteractions(executor, metadata, coordinator); + verify(client, never()).setDatabase(any()); + } finally { + Whitebox.setInternalState( + TreeSchemaAutoCreatorAndVerifier.class, + "CONFIG_NODE_CLIENT_MANAGER", + originalClientManager); + IoTDBDescriptor.getInstance().getConfig().setAutoCreateSchemaEnabled(originalGlobalEnabled); + } + } + + @Test + public void testAsyncTableSqlLoadPreservesSessionSchemaCreationSetting() throws Exception { + assertAsyncLoadPreservesSessionSchemaCreationSetting(true); + } + + @Test + public void testAsyncTreeSqlLoadPreservesSessionSchemaCreationSetting() throws Exception { + assertAsyncLoadPreservesSessionSchemaCreationSetting(false); + } + + private void assertAsyncLoadPreservesSessionSchemaCreationSetting(final boolean tableModelRequest) + throws Exception { + final AtomicReference> handedOffAttributes = new AtomicReference<>(); + PowerMockito.mockStatic(LoadUtil.class); + when(LoadUtil.loadTsFileAsyncToActiveDir(anyList(), anyMap(), anyBoolean())) + .thenAnswer( + invocation -> { + handedOffAttributes.set(new HashMap<>(invocation.getArgument(1))); + return true; + }); + final File file = temporaryFolder.newFile("async-schema.tsfile"); + final File pendingDir = temporaryFolder.newFolder("pending"); + final boolean originalGlobalEnabled = + IoTDBDescriptor.getInstance().getConfig().isAutoCreateSchemaEnabled(); + try { + for (final boolean globalEnabled : new boolean[] {true, false}) { + IoTDBDescriptor.getInstance().getConfig().setAutoCreateSchemaEnabled(globalEnabled); + for (final boolean sessionEnabled : new boolean[] {false, true, false}) { + for (final boolean requested : new boolean[] {false, true, false}) { + handedOffAttributes.set(null); + final Map attributes = new HashMap<>(); + attributes.put(ASYNC_LOAD_KEY, "true"); + attributes.put(DATABASE_NAME_KEY, "db"); + if (!requested) { + attributes.put(AUTO_CREATE_SCHEMA_KEY, "false"); + } + final LoadTsFile tableStatement = + LoadTsFile.createUnchecked(null, file.getAbsolutePath(), attributes); + final LoadTsFileStatement treeStatement = + LoadTsFileStatement.createUnchecked(file.getAbsolutePath()); + treeStatement.setLoadAttributes(attributes); + final SessionInfo sessionInfo = + new SessionInfo( + 0, + new UserEntity(0, "root", ""), + ZoneOffset.UTC, + IoTDBConstant.ClientVersion.V_1_0, + "db", + tableModelRequest ? SqlDialect.TABLE : SqlDialect.TREE, + sessionEnabled); + final MPPQueryContext context = new MPPQueryContext(new QueryId("async_load_schema")); + context.setSession(sessionInfo); + try (final LoadTsFileAnalyzer analyzer = + tableModelRequest + ? new LoadTsFileAnalyzer(tableStatement, false, context) + : new LoadTsFileAnalyzer(treeStatement, false, context)) { + final Analysis analysis = new Analysis(); + analyzer.analyzeFileByFile(analysis); + Assert.assertTrue(analysis.isFinishQueryAfterAnalyze()); + Assert.assertFalse(analysis.isFailed()); + } + + Assert.assertNotNull(handedOffAttributes.get()); + // Reconstruct the request as the active loader does with its own internal session. + final File queuedFile = + new File( + ActiveLoadPathHelper.resolveTargetDir(pendingDir, handedOffAttributes.get()), + file.getName()); + final Map parsedAttributes = + ActiveLoadPathHelper.parseAttributes(queuedFile, pendingDir); + final LoadTsFileStatement activeStatement = + LoadTsFileStatement.createUnchecked(file.getAbsolutePath()); + ActiveLoadPathHelper.applyAttributesToStatement( + parsedAttributes, activeStatement, false); + Assert.assertEquals( + globalEnabled && requested && sessionEnabled, activeStatement.isAutoCreateSchema()); + Assert.assertEquals("db", activeStatement.getDatabase()); + Assert.assertSame(sessionInfo, context.getSession()); + Assert.assertEquals(sessionEnabled, sessionInfo.isEnableAutoCreateSchema()); + Assert.assertEquals(requested, tableStatement.isAutoCreateSchema()); + Assert.assertEquals(requested, treeStatement.isAutoCreateSchema()); + Assert.assertTrue(file.exists()); + } + } + } + } finally { + IoTDBDescriptor.getInstance().getConfig().setAutoCreateSchemaEnabled(originalGlobalEnabled); + } + } + + private void assertTableConversionPreservesAutoCreateSchema(final boolean miniFile) + throws Exception { + when(converter.convertForTableModel(any())) + .thenReturn(Optional.of(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()))); + for (final boolean tableModelRequest : new boolean[] {false, true}) { + for (final boolean requested : new boolean[] {false, true}) { + for (final boolean generatedByPipe : new boolean[] {false, true}) { + clearInvocations(converter); + final File file = + temporaryFolder.newFile( + "conversion-" + + tableModelRequest + + "-" + + requested + + "-" + + generatedByPipe + + ".tsfile"); + final LoadTsFile tableStatement = + LoadTsFile.createUnchecked( + null, + file.getAbsolutePath(), + requested + ? Collections.emptyMap() + : Collections.singletonMap("auto-create-schema", "false")) + .setDatabase("db"); + final LoadTsFileStatement treeStatement = + LoadTsFileStatement.createUnchecked(file.getAbsolutePath()); + treeStatement.setAutoCreateSchema(requested); + treeStatement.setDatabase("db"); + final MPPQueryContext context = + new MPPQueryContext(new QueryId("load_conversion_attributes")); + try (final LoadTsFileAnalyzer analyzer = + tableModelRequest + ? new LoadTsFileAnalyzer(tableStatement, generatedByPipe, context) + : new LoadTsFileAnalyzer(treeStatement, generatedByPipe, context)) { + final List tableModelFiles = + Whitebox.getInternalState(analyzer, "isTableModelTsFile"); + tableModelFiles.set(0, true); + Whitebox.setInternalState(analyzer, "isTableModelTsFileReliableIndex", 0); + if (miniFile) { + Assert.assertTrue( + (Boolean) Whitebox.invokeMethod(analyzer, "handleSingleMiniFile", 0)); + } else { + final Analysis analysis = new Analysis(); + Whitebox.invokeMethod( + analyzer, + "executeTabletConversionOnException", + (IAnalysis) analysis, + new LoadAnalyzeException("original analysis failure")); + Assert.assertTrue(analysis.isFinishQueryAfterAnalyze()); + } + final ArgumentCaptor statement = ArgumentCaptor.forClass(LoadTsFile.class); + verify(converter).convertForTableModel(statement.capture()); + Assert.assertEquals(requested, statement.getValue().isAutoCreateSchema()); + Assert.assertEquals("db", statement.getValue().getDatabase()); + } + } + } + } + } + private void mockModelReader(final boolean tableModel) throws Exception { final TsFileSequenceReader reader = mock(TsFileSequenceReader.class); when(reader.getTableSchemaMap()) diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileAnalyzerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileAnalyzerTest.java index b658b37dbf349..9ef5f7b7f624c 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileAnalyzerTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileAnalyzerTest.java @@ -19,9 +19,19 @@ package org.apache.iotdb.db.queryengine.plan.analyze.load; +import org.apache.iotdb.commons.audit.UserEntity; +import org.apache.iotdb.commons.conf.IoTDBConstant; +import org.apache.iotdb.commons.exception.SemanticException; +import org.apache.iotdb.commons.exception.table.TableNotExistsException; import org.apache.iotdb.commons.path.PartialPath; +import org.apache.iotdb.commons.queryengine.common.SessionInfo; +import org.apache.iotdb.commons.queryengine.common.SqlDialect; import org.apache.iotdb.commons.queryengine.plan.relational.metadata.ColumnSchema; +import org.apache.iotdb.commons.schema.table.TsTable; +import org.apache.iotdb.commons.schema.table.column.FieldColumnSchema; +import org.apache.iotdb.commons.schema.table.column.TagColumnSchema; import org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory; +import org.apache.iotdb.db.auth.AuthorityChecker; import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.exception.load.LoadAnalyzeException; import org.apache.iotdb.db.exception.load.LoadAnalyzeInvalidPathException; @@ -33,10 +43,17 @@ import org.apache.iotdb.db.queryengine.common.QueryId; import org.apache.iotdb.db.queryengine.common.schematree.ClusterSchemaTree; import org.apache.iotdb.db.queryengine.common.schematree.ISchemaTree; +import org.apache.iotdb.db.queryengine.plan.analyze.Analysis; +import org.apache.iotdb.db.queryengine.plan.analyze.lock.DataNodeSchemaLockManager; +import org.apache.iotdb.db.queryengine.plan.analyze.schema.ISchemaFetcher; +import org.apache.iotdb.db.queryengine.plan.relational.metadata.Metadata; 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.schemaengine.table.DataNodeTableCache; +import org.apache.iotdb.db.schemaengine.table.ITableCache; import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; import org.apache.iotdb.db.storageengine.dataregion.utils.TsFileTableSchemaRegistrationUtils; +import org.apache.iotdb.rpc.TSStatusCode; import org.apache.tsfile.enums.ColumnCategory; import org.apache.tsfile.enums.TSDataType; @@ -45,8 +62,11 @@ import org.apache.tsfile.file.metadata.TableSchema; import org.apache.tsfile.read.TsFileSequenceReader; import org.apache.tsfile.read.TsFileSequenceReaderTimeseriesMetadataIterator; +import org.apache.tsfile.read.common.Path; import org.apache.tsfile.read.common.type.TypeFactory; +import org.apache.tsfile.write.TsFileWriter; import org.apache.tsfile.write.chunk.AlignedChunkWriterImpl; +import org.apache.tsfile.write.record.Tablet; import org.apache.tsfile.write.schema.IMeasurementSchema; import org.apache.tsfile.write.schema.MeasurementSchema; import org.apache.tsfile.write.schema.Schema; @@ -55,18 +75,33 @@ import org.junit.Assert; import org.junit.Before; import org.junit.Test; +import org.mockito.ArgumentCaptor; import java.io.File; import java.lang.reflect.Field; import java.lang.reflect.InvocationTargetException; import java.lang.reflect.Method; import java.nio.file.Files; +import java.time.ZoneId; import java.util.Arrays; import java.util.Collections; +import java.util.HashMap; import java.util.HashSet; import java.util.List; import java.util.Map; import java.util.Set; +import java.util.concurrent.atomic.AtomicBoolean; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyBoolean; +import static org.mockito.ArgumentMatchers.anyList; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.atLeastOnce; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; public class LoadTsFileAnalyzerTest { @@ -83,6 +118,310 @@ public void tearDown() { IoTDBDescriptor.getInstance().getConfig().setDataNodeId(dataNodeId); } + @Test + public void testTableLoadDoesNotAutoCreateSchemaWhenRequestDisablesIt() throws Exception { + for (final boolean tableModelStatement : new boolean[] {true, false}) { + assertTableLoadDoesNotAutoCreateSchema(tableModelStatement, true, false, true); + } + } + + @Test + public void testTableLoadDoesNotAutoCreateSchemaWhenGlobalConfigDisablesIt() throws Exception { + for (final boolean tableModelStatement : new boolean[] {true, false}) { + assertTableLoadDoesNotAutoCreateSchema(tableModelStatement, false, true, true); + } + } + + @Test + public void testTableLoadDoesNotAutoCreateSchemaWhenSessionDisablesIt() throws Exception { + for (final boolean tableModelStatement : new boolean[] {true, false}) { + assertTableLoadDoesNotAutoCreateSchema(tableModelStatement, true, true, false); + } + } + + @Test + public void testTableTsFileAnalysisPreservesSessionAutoCreateSchemaSetting() throws Exception { + final boolean originalAutoCreateSchemaEnabled = + IoTDBDescriptor.getInstance().getConfig().isAutoCreateSchemaEnabled(); + final long originalTabletConversionThresholdBytes = + IoTDBDescriptor.getInstance().getConfig().getLoadTabletConversionThresholdBytes(); + final File tsFile = File.createTempFile("table-load-session-schema", ".tsfile"); + writeTableTsFileWithMixedDevices(tsFile); + + try { + IoTDBDescriptor.getInstance().getConfig().setAutoCreateSchemaEnabled(true); + IoTDBDescriptor.getInstance().getConfig().setLoadTabletConversionThresholdBytes(-1); + for (final boolean tableModelStatement : new boolean[] {true, false}) { + final String database = "load_session_schema_" + tableModelStatement; + final MPPQueryContext context = + createLoadQueryContext( + "load_session_schema_" + tableModelStatement, + database, + tableModelStatement ? SqlDialect.TABLE : SqlDialect.TREE, + false); + final LoadTsFile tableStatement = + LoadTsFile.createUnchecked(null, tsFile.getAbsolutePath(), Collections.emptyMap()) + .setDatabase(database); + tableStatement.setAutoCreateDatabase(false); + final LoadTsFileStatement treeStatement = + LoadTsFileStatement.createUnchecked(tsFile.getAbsolutePath()); + treeStatement.setDatabase(database); + treeStatement.setAutoCreateSchema(true); + treeStatement.setAutoCreateDatabase(false); + final boolean[] schemaCreationEnabledDuringValidation = {true}; + final SqlDialect[] validationDialect = {SqlDialect.TREE}; + final Metadata metadata = mock(Metadata.class); + when(metadata.validateTableHeaderSchema4TsFile( + anyString(), + any( + org.apache.iotdb.commons.queryengine.plan.relational.metadata.TableSchema + .class), + any(MPPQueryContext.class), + anyBoolean(), + anyBoolean(), + any(AtomicBoolean.class))) + .thenAnswer( + invocation -> { + final MPPQueryContext validationContext = invocation.getArgument(2); + schemaCreationEnabledDuringValidation[0] = + validationContext.getSession().isEnableAutoCreateSchema(); + validationDialect[0] = validationContext.getSession().getSqlDialect(); + return java.util.Optional.empty(); + }); + + try (final LoadTsFileAnalyzer analyzer = + tableModelStatement + ? new LoadTsFileAnalyzer(tableStatement, false, context) + : new LoadTsFileAnalyzer(treeStatement, false, context)) { + final Field metadataField = LoadTsFileAnalyzer.class.getDeclaredField("metadata"); + metadataField.setAccessible(true); + metadataField.set(analyzer, metadata); + final Method analyzeSingleTsFileMethod = + LoadTsFileAnalyzer.class.getDeclaredMethod( + "analyzeSingleTsFile", File.class, int.class); + analyzeSingleTsFileMethod.setAccessible(true); + final InvocationTargetException exception = + Assert.assertThrows( + InvocationTargetException.class, + () -> analyzeSingleTsFileMethod.invoke(analyzer, tsFile, 0)); + Assert.assertTrue(exception.getCause() instanceof LoadAnalyzeException); + } finally { + DataNodeSchemaLockManager.getInstance().releaseReadLock(context); + } + + Assert.assertFalse(schemaCreationEnabledDuringValidation[0]); + Assert.assertEquals(SqlDialect.TABLE, validationDialect[0]); + Assert.assertFalse(context.getSession().isEnableAutoCreateSchema()); + } + } finally { + IoTDBDescriptor.getInstance() + .getConfig() + .setAutoCreateSchemaEnabled(originalAutoCreateSchemaEnabled); + IoTDBDescriptor.getInstance() + .getConfig() + .setLoadTabletConversionThresholdBytes(originalTabletConversionThresholdBytes); + Assert.assertTrue(tsFile.delete()); + } + } + + @Test + public void testTreeTsFileAnalysisPreservesSessionAutoCreateSchemaSetting() throws Exception { + final boolean originalAutoCreateSchemaEnabled = + IoTDBDescriptor.getInstance().getConfig().isAutoCreateSchemaEnabled(); + final long originalTabletConversionThresholdBytes = + IoTDBDescriptor.getInstance().getConfig().getLoadTabletConversionThresholdBytes(); + final File tsFile = File.createTempFile("tree-load-session-schema", ".tsfile"); + final List schemas = + Collections.singletonList(new MeasurementSchema("s1", TSDataType.INT64)); + try (TsFileWriter writer = new TsFileWriter(tsFile)) { + writer.registerTimeseries(new Path("root.load_session_schema.d1"), schemas.get(0)); + final Tablet tablet = new Tablet("root.load_session_schema.d1", schemas, 1); + tablet.addTimestamp(0, 1); + tablet.addValue("s1", 0, 1L); + writer.writeTree(tablet); + } + + try { + IoTDBDescriptor.getInstance().getConfig().setAutoCreateSchemaEnabled(true); + IoTDBDescriptor.getInstance().getConfig().setLoadTabletConversionThresholdBytes(-1); + final MPPQueryContext context = + spy( + createLoadQueryContext( + "tree_load_session_schema", "root.load_session_schema", SqlDialect.TREE, false)); + final LoadTsFileStatement statement = + LoadTsFileStatement.createUnchecked(tsFile.getAbsolutePath()); + statement.setAutoCreateSchema(true); + statement.setAutoCreateDatabase(false); + statement.setVerifySchema(true); + + final ISchemaFetcher schemaFetcher = mock(ISchemaFetcher.class); + final ClusterSchemaTree existingSchema = new ClusterSchemaTree(); + existingSchema.appendSingleMeasurement( + new PartialPath("root.load_session_schema.d1.s1"), + schemas.get(0), + null, + null, + null, + false); + when(schemaFetcher.fetchSchemaList(anyList(), anyList(), any(MPPQueryContext.class))) + .thenReturn(existingSchema); + + try (final LoadTsFileAnalyzer analyzer = new LoadTsFileAnalyzer(statement, false, context)) { + final Field schemaFetcherField = LoadTsFileAnalyzer.class.getDeclaredField("schemaFetcher"); + schemaFetcherField.setAccessible(true); + schemaFetcherField.set(analyzer, schemaFetcher); + final Analysis analysis = new Analysis(); + analyzer.analyzeFileByFile(analysis); + Assert.assertFalse(analysis.isFailed()); + } finally { + DataNodeSchemaLockManager.getInstance().releaseReadLock(context); + } + + verify(schemaFetcher).fetchSchemaList(anyList(), anyList(), any(MPPQueryContext.class)); + verify(schemaFetcher, never()) + .fetchSchemaListWithAutoCreate( + anyList(), + anyList(), + anyList(), + anyList(), + anyList(), + anyList(), + any(MPPQueryContext.class)); + final ArgumentCaptor analysisSessions = + ArgumentCaptor.forClass(SessionInfo.class); + verify(context, atLeastOnce()).setSession(analysisSessions.capture()); + for (final SessionInfo session : analysisSessions.getAllValues()) { + Assert.assertFalse(session.isEnableAutoCreateSchema()); + Assert.assertEquals(SqlDialect.TREE, session.getSqlDialect()); + } + Assert.assertFalse(context.getSession().isEnableAutoCreateSchema()); + } finally { + IoTDBDescriptor.getInstance() + .getConfig() + .setAutoCreateSchemaEnabled(originalAutoCreateSchemaEnabled); + IoTDBDescriptor.getInstance() + .getConfig() + .setLoadTabletConversionThresholdBytes(originalTabletConversionThresholdBytes); + Assert.assertTrue(tsFile.delete()); + } + } + + private void assertTableLoadDoesNotAutoCreateSchema( + final boolean tableModelStatement, + final boolean globalAutoCreateSchemaEnabled, + final boolean requestAutoCreateSchema, + final boolean sessionAutoCreateSchemaEnabled) + throws Exception { + final boolean originalAutoCreateSchemaEnabled = + IoTDBDescriptor.getInstance().getConfig().isAutoCreateSchemaEnabled(); + final boolean originalEnablePartialInsert = + IoTDBDescriptor.getInstance().getConfig().isEnablePartialInsert(); + final File tsFile = File.createTempFile("table-load-auto-create-schema", ".tsfile"); + final String database = "load_auto_create_schema_test"; + final MPPQueryContext context = + createTableQueryContext( + "load_auto_create_schema", database, sessionAutoCreateSchemaEnabled); + final ITableCache tableCache = DataNodeTableCache.getInstance(); + + try { + IoTDBDescriptor.getInstance() + .getConfig() + .setAutoCreateSchemaEnabled(globalAutoCreateSchemaEnabled); + IoTDBDescriptor.getInstance().getConfig().setEnablePartialInsert(false); + tableCache.invalid(database); + final TsTable existingTable = new TsTable("table1"); + existingTable.addColumnSchema(new TagColumnSchema("tag1", TSDataType.STRING)); + existingTable.addColumnSchema(new FieldColumnSchema("s1", TSDataType.INT64)); + tableCache.preUpdateTable(database, existingTable, null); + tableCache.commitUpdateTable(database, "table1", null, null); + + final LoadTsFile tableStatement = + LoadTsFile.createUnchecked( + null, + tsFile.getAbsolutePath(), + requestAutoCreateSchema + ? Collections.emptyMap() + : Collections.singletonMap("auto-create-schema", "false")) + .setDatabase(database); + tableStatement.setAutoCreateDatabase(true); + final LoadTsFileStatement treeStatement = + LoadTsFileStatement.createUnchecked(tsFile.getAbsolutePath()); + treeStatement.setAutoCreateSchema(requestAutoCreateSchema); + treeStatement.setAutoCreateDatabase(true); + + try (final LoadTsFileAnalyzer analyzer = + tableModelStatement + ? new LoadTsFileAnalyzer(tableStatement, false, context) + : new LoadTsFileAnalyzer(treeStatement, false, context)) { + final Method getOrCreateTableSchemaCacheMethod = + LoadTsFileAnalyzer.class.getDeclaredMethod("getOrCreateTableSchemaCache"); + getOrCreateTableSchemaCacheMethod.setAccessible(true); + final LoadTsFileTableSchemaCache schemaCache = + (LoadTsFileTableSchemaCache) getOrCreateTableSchemaCacheMethod.invoke(analyzer); + schemaCache.setDatabase(database); + + schemaCache.setTableSchemaMap( + new HashMap<>( + Collections.singletonMap( + "missing_table", createTsFileTableSchema("missing_table", false)))); + final SemanticException missingTableException = + Assert.assertThrows( + SemanticException.class, + () -> schemaCache.createTableAndDatabaseIfNecessary("missing_table")); + Assert.assertTrue(missingTableException.getCause() instanceof TableNotExistsException); + Assert.assertNull(tableCache.getTableInCache(database, "missing_table")); + + schemaCache.setTableSchemaMap( + new HashMap<>( + Collections.singletonMap("table1", createTsFileTableSchema("table1", true)))); + final SemanticException missingColumnException = + Assert.assertThrows( + SemanticException.class, + () -> schemaCache.createTableAndDatabaseIfNecessary("table1")); + Assert.assertEquals( + TSStatusCode.COLUMN_NOT_EXISTS.getStatusCode(), missingColumnException.getErrorCode()); + Assert.assertNull(tableCache.getTableInCache(database, "table1").getColumnSchema("s2")); + + // Existing partial-insert behavior still applies without creating the missing field. + IoTDBDescriptor.getInstance().getConfig().setEnablePartialInsert(true); + schemaCache.setTableSchemaMap( + new HashMap<>( + Collections.singletonMap("table1", createTsFileTableSchema("table1", true)))); + schemaCache.createTableAndDatabaseIfNecessary("table1"); + Assert.assertNull(tableCache.getTableInCache(database, "table1").getColumnSchema("s2")); + + schemaCache.setTableSchemaMap( + new HashMap<>( + Collections.singletonMap("table1", createTsFileTableSchema("table1", false)))); + schemaCache.createTableAndDatabaseIfNecessary("table1"); + } + } finally { + DataNodeSchemaLockManager.getInstance().releaseReadLock(context); + tableCache.invalid(database); + IoTDBDescriptor.getInstance() + .getConfig() + .setAutoCreateSchemaEnabled(originalAutoCreateSchemaEnabled); + IoTDBDescriptor.getInstance().getConfig().setEnablePartialInsert(originalEnablePartialInsert); + Assert.assertTrue(tsFile.delete()); + } + } + + private TableSchema createTsFileTableSchema(final String tableName, final boolean extraField) { + return new TableSchema( + tableName, + extraField + ? Arrays.asList( + new MeasurementSchema("tag1", TSDataType.STRING), + new MeasurementSchema("s1", TSDataType.INT64), + new MeasurementSchema("s2", TSDataType.DOUBLE)) + : Arrays.asList( + new MeasurementSchema("tag1", TSDataType.STRING), + new MeasurementSchema("s1", TSDataType.INT64)), + extraField + ? Arrays.asList(ColumnCategory.TAG, ColumnCategory.FIELD, ColumnCategory.FIELD) + : Arrays.asList(ColumnCategory.TAG, ColumnCategory.FIELD)); + } + @Test public void testAnalyzeSingleTableFileShouldNotCountTimestampInPointCount() throws Exception { final File tsFile = new File("load-table-mixed-null-device.tsfile"); @@ -539,6 +878,29 @@ private boolean containsDevice(final Set devices, final String... exp .anyMatch(device -> Arrays.equals(device.getSegments(), expectedSegments)); } + private MPPQueryContext createTableQueryContext( + final String queryId, final String database, final boolean enableAutoCreateSchema) { + return createLoadQueryContext(queryId, database, SqlDialect.TABLE, enableAutoCreateSchema); + } + + private MPPQueryContext createLoadQueryContext( + final String queryId, + final String database, + final SqlDialect dialect, + final boolean enableAutoCreateSchema) { + final MPPQueryContext context = new MPPQueryContext(new QueryId(queryId)); + context.setSession( + new SessionInfo( + 0, + new UserEntity(AuthorityChecker.SUPER_USER_ID, AuthorityChecker.SUPER_USER, ""), + ZoneId.systemDefault(), + IoTDBConstant.ClientVersion.V_1_0, + database, + dialect, + enableAutoCreateSchema)); + return context; + } + private static class TrackingLoadTsFileTableSchemaCache extends LoadTsFileTableSchemaCache { private final Set> verifiedDevices = new HashSet<>(); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/analyze/schema/TreeSchemaAutoCreationPolicyTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/analyze/schema/TreeSchemaAutoCreationPolicyTest.java new file mode 100644 index 0000000000000..560191c72ce12 --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/analyze/schema/TreeSchemaAutoCreationPolicyTest.java @@ -0,0 +1,207 @@ +/* + * 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.queryengine.plan.analyze.schema; + +import org.apache.iotdb.commons.audit.UserEntity; +import org.apache.iotdb.commons.conf.IoTDBConstant; +import org.apache.iotdb.commons.path.MeasurementPath; +import org.apache.iotdb.commons.path.PartialPath; +import org.apache.iotdb.commons.queryengine.common.SessionInfo; +import org.apache.iotdb.commons.queryengine.common.SqlDialect; +import org.apache.iotdb.commons.schema.template.Template; +import org.apache.iotdb.db.conf.IoTDBDescriptor; +import org.apache.iotdb.db.queryengine.common.MPPQueryContext; +import org.apache.iotdb.db.queryengine.common.QueryId; +import org.apache.iotdb.db.queryengine.common.schematree.ClusterSchemaTree; +import org.apache.iotdb.db.queryengine.common.schematree.IMeasurementSchemaInfo; +import org.apache.iotdb.db.queryengine.plan.relational.metadata.fetcher.cache.TreeDeviceSchemaCacheManager; + +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.utils.Pair; +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; + +import java.time.ZoneOffset; +import java.util.Collections; +import java.util.List; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyList; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.ArgumentMatchers.same; +import static org.mockito.Mockito.clearInvocations; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyZeroInteractions; +import static org.mockito.Mockito.when; + +public class TreeSchemaAutoCreationPolicyTest { + + private boolean originalAutoCreateSchema; + private TreeDeviceSchemaCacheManager cache; + private AutoCreateSchemaExecutor creator; + private ClusterSchemaFetchExecutor fetcher; + private ISchemaComputationWithAutoCreation computation; + private ClusterSchemaTree fetchedSchema; + private PartialPath device; + + @Before + public void setUp() throws Exception { + originalAutoCreateSchema = + IoTDBDescriptor.getInstance().getConfig().isAutoCreateSchemaEnabled(); + IoTDBDescriptor.getInstance().getConfig().setAutoCreateSchemaEnabled(true); + cache = mock(TreeDeviceSchemaCacheManager.class); + creator = mock(AutoCreateSchemaExecutor.class); + fetcher = mock(ClusterSchemaFetchExecutor.class); + computation = mock(ISchemaComputationWithAutoCreation.class); + device = new PartialPath("root.load_policy.d1"); + when(computation.getDevicePath()).thenReturn(device); + when(computation.getMeasurements()).thenReturn(new String[] {"s1"}); + when(computation.getDataType(0)).thenReturn(TSDataType.INT64); + when(cache.computeWithoutTemplate(computation)).thenReturn(Collections.singletonList(0)); + when(cache.computeWithTemplate(computation)).thenReturn(Collections.singletonList(0)); + when(cache.computeSourceOfLogicalView(computation)) + .thenReturn(new Pair<>(Collections.emptyList(), Collections.emptyList())); + fetchedSchema = new ClusterSchemaTree(); + when(fetcher.fetchSchemaOfOneDevice(any(), any(), anyList(), any())).thenReturn(fetchedSchema); + when(fetcher.fetchSchemaOfMultiDevices(anyList(), anyList(), anyList(), anyList(), any())) + .thenReturn(fetchedSchema); + } + + @After + public void tearDown() { + IoTDBDescriptor.getInstance().getConfig().setAutoCreateSchemaEnabled(originalAutoCreateSchema); + } + + @Test + public void testNormalSingleDeviceSchemaCreationPolicy() { + final NormalSchemaFetcher schemaFetcher = new NormalSchemaFetcher(cache, creator, fetcher); + for (final boolean allowed : new boolean[] {false, true, false}) { + clearInvocations(creator); + final MPPQueryContext context = context(allowed); + Assert.assertEquals( + Collections.singletonList(0), + schemaFetcher.processNormalTimeSeries(computation, context)); + if (allowed) { + verify(creator) + .autoCreateTimeSeries( + any(), eq(device), anyList(), any(), any(), eq(false), same(context)); + } else { + verifyZeroInteractions(creator); + } + } + } + + @Test + public void testNormalMultiDeviceSchemaCreationPolicy() { + final NormalSchemaFetcher schemaFetcher = new NormalSchemaFetcher(cache, creator, fetcher); + for (final boolean allowed : new boolean[] {false, true, false}) { + clearInvocations(creator, computation); + final MPPQueryContext context = context(allowed); + schemaFetcher.processNormalTimeSeries(Collections.singletonList(computation), context); + if (allowed) { + verify(creator) + .autoCreateTimeSeries( + any(), + anyList(), + anyList(), + anyList(), + anyList(), + anyList(), + anyList(), + same(context)); + } else { + verifyZeroInteractions(creator); + } + verify(computation).computeMeasurement(0, null); + } + } + + @Test + public void testExistingSchemaRemainsUsableWhenCreationIsDisabled() throws Exception { + fetchedSchema.appendSingleMeasurementPath( + new MeasurementPath("root.load_policy.d1.s1", TSDataType.INT64)); + final NormalSchemaFetcher schemaFetcher = new NormalSchemaFetcher(cache, creator, fetcher); + Assert.assertTrue(schemaFetcher.processNormalTimeSeries(computation, context(false)).isEmpty()); + verify(computation).computeMeasurement(eq(0), any(IMeasurementSchemaInfo.class)); + verifyZeroInteractions(creator); + } + + @Test + public void testTemplateSingleDeviceSchemaCreationPolicy() { + final TemplateSchemaFetcher schemaFetcher = new TemplateSchemaFetcher(cache, creator, fetcher); + final Template template = mock(Template.class); + when(template.getName()).thenReturn("load_template"); + when(template.getId()).thenReturn(7); + for (final boolean allowed : new boolean[] {false, true, false}) { + clearInvocations(creator); + final MPPQueryContext context = context(allowed); + Assert.assertEquals( + Collections.singletonList(0), + schemaFetcher.processTemplateTimeSeries( + new Pair<>(template, device), computation, context)); + if (allowed) { + verify(creator) + .autoExtendTemplate(eq("load_template"), anyList(), anyList(), same(context)); + verify(creator).autoActivateTemplate(any(), eq(device), eq(7), same(context)); + } else { + verifyZeroInteractions(creator); + } + } + } + + @Test + public void testTemplateMultiDeviceSchemaCreationPolicy() { + final TemplateSchemaFetcher schemaFetcher = new TemplateSchemaFetcher(cache, creator, fetcher); + final Template template = mock(Template.class); + when(template.getName()).thenReturn("load_template"); + final List> templateInfo = + Collections.singletonList(new Pair<>(template, device)); + for (final boolean allowed : new boolean[] {false, true, false}) { + clearInvocations(creator, computation); + final MPPQueryContext context = context(allowed); + schemaFetcher.processTemplateTimeSeries( + templateInfo, Collections.singletonList(computation), context); + if (allowed) { + verify(creator).autoExtendTemplate(any(), same(context)); + verify(creator).autoActivateTemplate(any(), anyList(), anyList(), same(context)); + } else { + verifyZeroInteractions(creator); + } + verify(computation).computeMeasurement(0, null); + } + } + + private static MPPQueryContext context(final boolean allowed) { + final MPPQueryContext context = new MPPQueryContext(new QueryId("load_policy")); + context.setSession( + new SessionInfo( + 0, + new UserEntity(0, "root", ""), + ZoneOffset.UTC, + IoTDBConstant.ClientVersion.V_1_0, + null, + SqlDialect.TREE, + allowed)); + return context; + } +} diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadFallbackHandlerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadFallbackHandlerTest.java index b765405277a4f..97aee5857b114 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadFallbackHandlerTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadFallbackHandlerTest.java @@ -20,14 +20,22 @@ package org.apache.iotdb.db.queryengine.plan.scheduler.load; import org.apache.iotdb.common.rpc.thrift.TSStatus; +import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId; import org.apache.iotdb.db.i18n.DataNodeQueryMessages; import org.apache.iotdb.db.queryengine.common.MPPQueryContext; import org.apache.iotdb.db.queryengine.common.QueryId; import org.apache.iotdb.db.queryengine.execution.QueryState; import org.apache.iotdb.db.queryengine.execution.QueryStateMachine; +import org.apache.iotdb.db.queryengine.plan.analyze.IAnalysis; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.load.LoadSingleTsFileNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.load.LoadTsFileNode; +import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.LoadTsFile; +import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.PipeEnriched; +import org.apache.iotdb.db.queryengine.plan.statement.crud.LoadTsFileStatement; +import org.apache.iotdb.db.queryengine.plan.statement.pipe.PipeEnrichedStatement; import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; import org.apache.iotdb.db.storageengine.load.converter.LoadTsFileDataTypeConverter; +import org.apache.iotdb.db.storageengine.load.metrics.LoadTsFileCostMetricsSet; import org.apache.iotdb.rpc.RpcUtils; import org.apache.iotdb.rpc.TSStatusCode; @@ -38,6 +46,7 @@ import org.junit.Test; import org.junit.rules.TemporaryFolder; import org.junit.runner.RunWith; +import org.mockito.ArgumentCaptor; import org.powermock.api.mockito.PowerMockito; import org.powermock.core.classloader.annotations.PowerMockIgnore; import org.powermock.core.classloader.annotations.PrepareForTest; @@ -51,7 +60,9 @@ import java.util.Optional; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.clearInvocations; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @RunWith(PowerMockRunner.class) @@ -134,6 +145,134 @@ public void testSuccessfulConversionsStillFinishLoad() throws Exception { Assert.assertTrue(failedIndexes.isEmpty()); } + @Test + public void testTableFallbackPreservesRequestSchemaCreationSetting() throws Exception { + assertTableConversionPreservesRequestSchemaCreationSetting(false); + } + + @Test + public void testTableRewritePreservesRequestSchemaCreationSetting() throws Exception { + assertTableConversionPreservesRequestSchemaCreationSetting(true); + } + + @Test + public void testTreeFallbackPreservesRequestSchemaCreationSetting() throws Exception { + when(converter.convertForTreeModel(any())).thenReturn(Optional.of(RpcUtils.SUCCESS_STATUS)); + final File file = temporaryFolder.newFile("tree-schema.tsfile"); + for (final boolean tableModelRequest : new boolean[] {false, true}) { + for (final boolean generatedByPipe : new boolean[] {false, true}) { + for (final boolean requested : new boolean[] {false, true, false}) { + clearInvocations(converter); + final LoadSingleTsFileNode node = + splitLoadNode(file, tableModelRequest, generatedByPipe, requested, false); + final List failedIndexes = new ArrayList<>(Collections.singletonList(0)); + final QueryStateMachine stateMachine = newStateMachine(); + new LoadFallbackHandler( + mock(MPPQueryContext.class), + generatedByPipe, + Collections.singletonList(node), + failedIndexes, + stateMachine) + .convertFailedTsFilesToTablets(); + Assert.assertEquals(QueryState.FINISHED, stateMachine.getState()); + Assert.assertTrue(failedIndexes.isEmpty()); + final ArgumentCaptor statement = + ArgumentCaptor.forClass(LoadTsFileStatement.class); + verify(converter).convertForTreeModel(statement.capture()); + Assert.assertEquals(requested, statement.getValue().isAutoCreateSchema()); + Assert.assertEquals(generatedByPipe, statement.getValue().isGeneratedByPipe()); + Assert.assertTrue(statement.getValue().isDeleteAfterLoad()); + Assert.assertEquals(Collections.singletonList(file), statement.getValue().getTsFiles()); + Assert.assertTrue(file.exists()); + } + } + } + } + + private void assertTableConversionPreservesRequestSchemaCreationSetting(final boolean rewrite) + throws Exception { + when(converter.convertForTableModel(any())).thenReturn(Optional.of(RpcUtils.SUCCESS_STATUS)); + final File file = temporaryFolder.newFile("table-schema.tsfile"); + for (final boolean tableModelRequest : new boolean[] {false, true}) { + for (final boolean generatedByPipe : new boolean[] {false, true}) { + for (final boolean requested : new boolean[] {false, true, false}) { + clearInvocations(converter); + final LoadSingleTsFileNode node = + splitLoadNode(file, tableModelRequest, generatedByPipe, requested, true); + if (rewrite) { + Assert.assertTrue( + new TabletRewriteLoadStrategy( + converter, generatedByPipe, mock(LoadTsFileCostMetricsSet.class)) + .execute(node)); + } else { + final List failedIndexes = new ArrayList<>(Collections.singletonList(0)); + final QueryStateMachine stateMachine = newStateMachine(); + new LoadFallbackHandler( + mock(MPPQueryContext.class), + generatedByPipe, + Collections.singletonList(node), + failedIndexes, + stateMachine) + .convertFailedTsFilesToTablets(); + Assert.assertEquals(QueryState.FINISHED, stateMachine.getState()); + Assert.assertTrue(failedIndexes.isEmpty()); + } + final ArgumentCaptor statement = ArgumentCaptor.forClass(LoadTsFile.class); + verify(converter).convertForTableModel(statement.capture()); + Assert.assertEquals(requested, statement.getValue().isAutoCreateSchema()); + Assert.assertEquals("test", statement.getValue().getDatabase()); + Assert.assertEquals(generatedByPipe, statement.getValue().isGeneratedByPipe()); + Assert.assertEquals(!rewrite, statement.getValue().isDeleteAfterLoad()); + Assert.assertTrue(file.exists()); + } + } + } + } + + private LoadSingleTsFileNode splitLoadNode( + final File file, + final boolean tableModelRequest, + final boolean generatedByPipe, + final boolean requested, + final boolean tableModelFile) + throws Exception { + final IAnalysis analysis; + if (tableModelRequest) { + final LoadTsFile statement = + LoadTsFile.createUnchecked( + null, + file.getAbsolutePath(), + requested + ? Collections.emptyMap() + : Collections.singletonMap("auto-create-schema", "false")) + .setDatabase("test") + .setDeleteAfterLoad(true); + statement.addWritePointCount(1); + final org.apache.iotdb.db.queryengine.plan.relational.analyzer.Analysis tableAnalysis = + new org.apache.iotdb.db.queryengine.plan.relational.analyzer.Analysis( + generatedByPipe ? new PipeEnriched(statement) : statement, Collections.emptyMap()); + analysis = tableAnalysis; + } else { + final LoadTsFileStatement statement = + LoadTsFileStatement.createUnchecked(file.getAbsolutePath()).setDeleteAfterLoad(true); + statement.setAutoCreateSchema(requested); + statement.addWritePointCount(1); + final org.apache.iotdb.db.queryengine.plan.analyze.Analysis treeAnalysis = + new org.apache.iotdb.db.queryengine.plan.analyze.Analysis(); + treeAnalysis.setRealStatement( + generatedByPipe ? new PipeEnrichedStatement(statement) : statement); + analysis = treeAnalysis; + } + final LoadTsFileNode loadNode = + new LoadTsFileNode( + new PlanNodeId("load_schema"), + Collections.singletonList(new TsFileResource(file)), + Collections.singletonList(tableModelFile), + tableModelFile ? "test" : "root.test", + false); + return (LoadSingleTsFileNode) loadNode.splitByPartition(analysis).get(0); + } + private LoadSingleTsFileNode newNode(final String name, final boolean tableModel) throws Exception { final File file = temporaryFolder.newFile(name); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/converter/LoadTsFileDataTypeConverterSchemaTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/converter/LoadTsFileDataTypeConverterSchemaTest.java new file mode 100644 index 0000000000000..86468fd9edd65 --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/converter/LoadTsFileDataTypeConverterSchemaTest.java @@ -0,0 +1,349 @@ +/* + * 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.storageengine.load.converter; + +import org.apache.iotdb.common.rpc.thrift.TSStatus; +import org.apache.iotdb.commons.audit.UserEntity; +import org.apache.iotdb.commons.conf.IoTDBConstant; +import org.apache.iotdb.commons.queryengine.common.SessionInfo; +import org.apache.iotdb.commons.queryengine.common.SqlDialect; +import org.apache.iotdb.db.auth.AuthorityChecker; +import org.apache.iotdb.db.conf.IoTDBDescriptor; +import org.apache.iotdb.db.protocol.session.IClientSession; +import org.apache.iotdb.db.protocol.session.InternalClientSession; +import org.apache.iotdb.db.protocol.session.SessionManager; +import org.apache.iotdb.db.queryengine.common.MPPQueryContext; +import org.apache.iotdb.db.queryengine.common.QueryId; +import org.apache.iotdb.db.queryengine.plan.Coordinator; +import org.apache.iotdb.db.queryengine.plan.analyze.IPartitionFetcher; +import org.apache.iotdb.db.queryengine.plan.analyze.lock.DataNodeSchemaLockManager; +import org.apache.iotdb.db.queryengine.plan.analyze.schema.ISchemaFetcher; +import org.apache.iotdb.db.queryengine.plan.execution.ExecutionResult; +import org.apache.iotdb.db.queryengine.plan.relational.metadata.Metadata; +import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.LoadTsFile; +import org.apache.iotdb.db.queryengine.plan.relational.sql.parser.SqlParser; +import org.apache.iotdb.db.queryengine.plan.statement.Statement; +import org.apache.iotdb.db.queryengine.plan.statement.crud.LoadTsFileStatement; +import org.apache.iotdb.rpc.TSStatusCode; + +import org.apache.tsfile.enums.ColumnCategory; +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.file.metadata.StringArrayDeviceID; +import org.apache.tsfile.file.metadata.TableSchema; +import org.apache.tsfile.read.common.Path; +import org.apache.tsfile.write.TsFileWriter; +import org.apache.tsfile.write.chunk.AlignedChunkWriterImpl; +import org.apache.tsfile.write.record.Tablet; +import org.apache.tsfile.write.schema.IMeasurementSchema; +import org.apache.tsfile.write.schema.MeasurementSchema; +import org.apache.tsfile.write.schema.Schema; +import org.apache.tsfile.write.writer.TsFileIOWriter; +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; +import org.junit.runner.RunWith; +import org.powermock.api.mockito.PowerMockito; +import org.powermock.core.classloader.annotations.PowerMockIgnore; +import org.powermock.core.classloader.annotations.PrepareForTest; +import org.powermock.modules.junit4.PowerMockRunner; + +import java.io.File; +import java.time.ZoneOffset; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.Optional; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyBoolean; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +@RunWith(PowerMockRunner.class) +@PowerMockIgnore({"com.sun.org.apache.xerces.*", "javax.xml.*", "org.xml.*", "javax.management.*"}) +@PrepareForTest({Coordinator.class, AuthorityChecker.class}) +public class LoadTsFileDataTypeConverterSchemaTest { + + @Rule public final TemporaryFolder temporaryFolder = new TemporaryFolder(); + + private final List conversionSessions = new ArrayList<>(); + private final List treeConversionSessions = new ArrayList<>(); + private InternalClientSession session; + private int originalDataNodeId; + private File tsFile; + private File treeTsFile; + + @Before + public void setUp() throws Exception { + originalDataNodeId = IoTDBDescriptor.getInstance().getConfig().getDataNodeId(); + IoTDBDescriptor.getInstance().getConfig().setDataNodeId(0); + session = new InternalClientSession("load_schema_conversion"); + session.setUserId(0); + session.setUsername("root"); + session.setClientVersion(IoTDBConstant.ClientVersion.V_1_0); + session.setZoneId(ZoneOffset.UTC); + session.setSqlDialect(SqlDialect.TREE); + SessionManager.getInstance().registerSession(session); + + PowerMockito.mockStatic(Coordinator.class); + PowerMockito.mockStatic(AuthorityChecker.class); + PowerMockito.when(AuthorityChecker.getUserId(anyString())).thenReturn(Optional.of(0L)); + final Coordinator coordinator = mock(Coordinator.class); + when(Coordinator.getInstance()).thenReturn(coordinator); + when(coordinator.executeForTableModel( + any(Statement.class), + any(SqlParser.class), + any(IClientSession.class), + anyLong(), + any(SessionInfo.class), + anyString(), + any(Metadata.class), + anyLong())) + .thenAnswer( + invocation -> { + conversionSessions.add(invocation.getArgument(4)); + return new ExecutionResult( + new QueryId("load_schema_conversion"), + new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode())); + }); + tsFile = new File(temporaryFolder.getRoot(), "table.tsfile"); + writeTableTsFile(); + when(coordinator.executeForTreeModel( + any(Statement.class), + anyLong(), + any(SessionInfo.class), + anyString(), + any(IPartitionFetcher.class), + any(ISchemaFetcher.class), + anyLong(), + anyBoolean(), + anyBoolean())) + .thenAnswer( + invocation -> { + treeConversionSessions.add(invocation.getArgument(2)); + return new ExecutionResult( + new QueryId("tree_load_schema_conversion"), + new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode())); + }); + treeTsFile = new File(temporaryFolder.getRoot(), "tree.tsfile"); + final List measurements = + Collections.singletonList(new MeasurementSchema("s1", TSDataType.INT64)); + try (final TsFileWriter writer = new TsFileWriter(treeTsFile)) { + writer.registerTimeseries(new Path("root.load_schema.d1"), measurements.get(0)); + final Tablet tablet = new Tablet("root.load_schema.d1", measurements, 1); + tablet.addTimestamp(0, -1); + tablet.addValue("s1", 0, 1L); + writer.writeTree(tablet); + } + } + + @After + public void tearDown() { + SessionManager.getInstance().removeCurrSession(); + IoTDBDescriptor.getInstance().getConfig().setDataNodeId(originalDataNodeId); + } + + @Test + public void testTableConversionHonorsRequestWithoutChangingClientSession() { + final LoadTsFileDataTypeConverter converter = + new LoadTsFileDataTypeConverter(new MPPQueryContext(new QueryId("load_schema")), false); + for (final boolean requested : new boolean[] {false, true, false}) { + assertConversionSchemaFlag(converter, requested, requested); + Assert.assertTrue(session.isEnableAutoCreateSchema()); + Assert.assertEquals(SqlDialect.TREE, session.getSqlDialect()); + Assert.assertSame(session, SessionManager.getInstance().getCurrSession()); + } + } + + @Test + public void testTableConversionPreservesDisabledClientSessionSchemaCreation() { + session.setEnableAutoCreateSchema(false); + final LoadTsFileDataTypeConverter converter = + new LoadTsFileDataTypeConverter(new MPPQueryContext(new QueryId("load_schema")), false); + assertConversionSchemaFlag(converter, true, false); + Assert.assertFalse(session.isEnableAutoCreateSchema()); + } + + @Test + public void testTableConversionPreservesDisabledQuerySessionWithoutClientSession() { + SessionManager.getInstance().removeCurrSession(); + final MPPQueryContext context = new MPPQueryContext(new QueryId("load_schema")); + context.setSession( + new SessionInfo( + session.getId(), + new UserEntity(session.getUserId(), session.getUsername(), session.getClientAddress()), + ZoneOffset.UTC, + session.getClientVersion(), + "db", + SqlDialect.TABLE, + false)); + final LoadTsFileDataTypeConverter converter = new LoadTsFileDataTypeConverter(context, false); + + assertConversionFlagWithoutClientSession(converter); + + Assert.assertNull(SessionManager.getInstance().getCurrSession()); + } + + @Test + public void testTreeConversionHonorsRequestWithoutChangingClientSession() throws Exception { + for (final boolean generatedByPipe : new boolean[] {false, true}) { + final MPPQueryContext context = new MPPQueryContext(new QueryId("tree_load_schema")); + final LoadTsFileDataTypeConverter converter = + new LoadTsFileDataTypeConverter(context, generatedByPipe); + try { + for (final boolean requested : new boolean[] {false, true, false}) { + assertTreeConversionSchemaFlag(converter, requested, requested); + Assert.assertTrue(session.isEnableAutoCreateSchema()); + Assert.assertEquals(SqlDialect.TREE, session.getSqlDialect()); + Assert.assertSame(session, SessionManager.getInstance().getCurrSession()); + } + } finally { + DataNodeSchemaLockManager.getInstance().releaseReadLock(context); + } + } + } + + @Test + public void testTreeConversionPreservesDisabledClientSession() throws Exception { + session.setEnableAutoCreateSchema(false); + final MPPQueryContext context = new MPPQueryContext(new QueryId("tree_load_schema")); + try { + assertTreeConversionSchemaFlag(new LoadTsFileDataTypeConverter(context, false), true, false); + Assert.assertFalse(session.isEnableAutoCreateSchema()); + } finally { + DataNodeSchemaLockManager.getInstance().releaseReadLock(context); + } + } + + @Test + public void testTreeConversionPreservesDisabledQuerySession() throws Exception { + final MPPQueryContext context = new MPPQueryContext(new QueryId("tree_load_schema")); + context.setSession( + new SessionInfo( + session.getId(), + new UserEntity(session.getUserId(), session.getUsername(), session.getClientAddress()), + ZoneOffset.UTC, + session.getClientVersion(), + null, + SqlDialect.TREE, + false)); + try { + final LoadTsFileDataTypeConverter converter = new LoadTsFileDataTypeConverter(context, false); + assertTreeConversionSchemaFlag(converter, true, false); + Assert.assertTrue(session.isEnableAutoCreateSchema()); + SessionManager.getInstance().removeCurrSession(); + assertTreeConversionSchemaFlag(converter, true, false); + Assert.assertNull(SessionManager.getInstance().getCurrSession()); + } finally { + DataNodeSchemaLockManager.getInstance().releaseReadLock(context); + } + } + + private void assertTreeConversionSchemaFlag( + final LoadTsFileDataTypeConverter converter, final boolean requested, final boolean expected) + throws Exception { + treeConversionSessions.clear(); + final LoadTsFileStatement statement = + LoadTsFileStatement.createUnchecked(treeTsFile.getAbsolutePath()); + statement.setAutoCreateSchema(requested); + final Optional status = converter.convertForTreeModel(statement); + Assert.assertTrue(status.isPresent()); + Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.get().getCode()); + Assert.assertFalse(treeConversionSessions.isEmpty()); + for (final SessionInfo conversionSession : treeConversionSessions) { + Assert.assertEquals(expected, conversionSession.isEnableAutoCreateSchema()); + Assert.assertEquals(SqlDialect.TREE, conversionSession.getSqlDialect()); + Assert.assertEquals("root", conversionSession.getUserName()); + } + Assert.assertEquals(requested, statement.isAutoCreateSchema()); + Assert.assertTrue(treeTsFile.exists()); + } + + private void assertConversionSchemaFlag( + final LoadTsFileDataTypeConverter converter, + final boolean requested, + final boolean expected) { + conversionSessions.clear(); + final LoadTsFile statement = + LoadTsFile.createUnchecked( + null, + tsFile.getAbsolutePath(), + requested + ? Collections.emptyMap() + : Collections.singletonMap("auto-create-schema", "false")) + .setDatabase("db"); + final Optional status = converter.convertForTableModel(statement); + Assert.assertTrue(status.isPresent()); + Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.get().getCode()); + Assert.assertFalse(conversionSessions.isEmpty()); + for (final SessionInfo conversionSession : conversionSessions) { + Assert.assertEquals(expected, conversionSession.isEnableAutoCreateSchema()); + Assert.assertEquals("db", conversionSession.getDatabaseName().orElse(null)); + Assert.assertEquals(SqlDialect.TABLE, conversionSession.getSqlDialect()); + Assert.assertEquals("root", conversionSession.getUserName()); + } + } + + private void assertConversionFlagWithoutClientSession( + final LoadTsFileDataTypeConverter converter) { + conversionSessions.clear(); + final LoadTsFile statement = + LoadTsFile.createUnchecked( + null, + tsFile.getAbsolutePath(), + Collections.singletonMap("auto-create-schema", "true")) + .setDatabase("db"); + final Optional status = converter.convertForTableModel(statement); + Assert.assertTrue(status.isPresent()); + Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.get().getCode()); + Assert.assertFalse(conversionSessions.isEmpty()); + for (final SessionInfo conversionSession : conversionSessions) { + Assert.assertFalse(conversionSession.isEnableAutoCreateSchema()); + } + } + + private void writeTableTsFile() throws Exception { + final List columns = + Arrays.asList( + new MeasurementSchema("tag1", TSDataType.STRING), + new MeasurementSchema("s1", TSDataType.INT64)); + final Schema schema = new Schema(); + schema.registerTableSchema( + new TableSchema( + "table1", columns, Arrays.asList(ColumnCategory.TAG, ColumnCategory.FIELD))); + try (final TsFileIOWriter writer = new TsFileIOWriter(tsFile)) { + writer.setSchema(schema); + writer.startChunkGroup(new StringArrayDeviceID(new String[] {"table1", "tagA"})); + final AlignedChunkWriterImpl chunkWriter = + new AlignedChunkWriterImpl(columns.subList(1, columns.size())); + chunkWriter.getTimeChunkWriter().write(1); + chunkWriter.getValueChunkWriterByIndex(0).write(1, 1L, false); + chunkWriter.writeToFileWriter(writer); + writer.endChunkGroup(); + writer.endFile(); + } + } +}