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 61fe99e9ea139..29905b6f88f29 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 @@ -49,6 +49,7 @@ import org.apache.tsfile.write.TsFileWriter; import org.apache.tsfile.write.record.Tablet; import org.apache.tsfile.write.schema.MeasurementSchema; +import org.awaitility.Awaitility; import org.junit.After; import org.junit.Assert; import org.junit.Before; @@ -62,6 +63,7 @@ import java.io.File; import java.io.IOException; import java.io.RandomAccessFile; +import java.nio.charset.StandardCharsets; import java.nio.file.Files; import java.sql.Connection; import java.sql.ResultSet; @@ -202,6 +204,121 @@ private boolean deleteDir() { return tmpDir.delete(); } + @Test + public void testQueryFallbackHonorsDisabledSchemaCreation() throws Exception { + final String database = "root.load.fallback_disabled_schema"; + final File source = new File(tmpDir, "712-0-0-0.tsfile"); + final long[] times = {-10001, -10000, -1, 0, 9999, 10000}; + final List schemas = + Arrays.asList( + new MeasurementSchema("s", TSDataType.INT64), + new MeasurementSchema("txt", TSDataType.TEXT)); + try (TsFileWriter writer = new TsFileWriter(source)) { + for (int device = 0; device < 2; device++) { + final String devicePath = database + ".d" + device; + writer.registerAlignedTimeseries(new Path(devicePath), schemas); + final Tablet tablet = new Tablet(devicePath, schemas, times.length); + for (int row = 0; row < times.length; row++) { + tablet.addTimestamp(row, times[row]); + tablet.addValue("s", row, device * 100L + row); + tablet.addValue("txt", row, row == 0 ? null : "d" + device + "_r" + row); + } + tablet.rowSize = times.length; + writer.writeAligned(tablet); + } + } + final TsFileResource resource = new TsFileResource(source); + for (int device = 0; device < 2; device++) { + final IDeviceID id = new PlainDeviceID(database + ".d" + device); + resource.updateStartTime(id, times[0]); + resource.updateEndTime(id, times[times.length - 1]); + } + resource.serialize(); + try (ModificationFile mods = resource.getModFile()) { + for (int device = 0; device < 2; device++) { + mods.write( + new Deletion( + new MeasurementPath(database + ".d" + device + ".s"), + Long.MAX_VALUE, + times[1], + times[1])); + } + } + try (TsFileSequenceReader reader = new TsFileSequenceReader(source.getAbsolutePath()); + RandomAccessFile bytes = new RandomAccessFile(source, "rw")) { + final long chunk = + reader + .getChunkMetadataList(new Path(new PlainDeviceID(database + ".d1"), "s", false)) + .get(0) + .getOffsetOfChunkHeader(); + reader.position(chunk); + reader.readChunkHeader(reader.readMarker()); + // Preserve readable metadata, but make both the piece reader and scan parser reject this + // chunk. Multiple time partitions prevent local loading from bypassing the piece reader. + bytes.seek(reader.position() - 2); + bytes.writeByte(0x7f); + } + final Map sourceFiles = new HashMap<>(); + for (String suffix : + new String[] {"", TsFileResource.RESOURCE_SUFFIX, ModificationFile.FILE_SUFFIX}) { + final File file = new File(source + suffix); + sourceFiles.put(file, Files.readAllBytes(file.toPath())); + } + final DataNodeWrapper node = EnvFactory.getEnv().getDataNodeWrapperList().get(0); + final File log = new File(node.getLogDirPath(), node.getId() + ".log"); + final long logOffset = Files.size(log.toPath()); + try (Connection connection = EnvFactory.getEnv().getConnectionWithSpecifiedDataNode(node); + Statement statement = connection.createStatement()) { + statement.execute("create database " + database); + try { + final SQLException failure = + Assert.assertThrows( + SQLException.class, + () -> + statement.execute( + "load '" + + source.getAbsolutePath() + + "' with ('database-level'='2','verify'='false'," + + "'auto-create-schema'='false','tablet-conversion-threshold'='-1'," + + "'on-success'='delete')")); + Assert.assertEquals(TSStatusCode.LOAD_FILE_ERROR.getStatusCode(), failure.getErrorCode()); + Assert.assertTrue(failure.getMessage(), failure.getMessage().contains(source.getName())); + final String fallbackMessage = + String.format( + "Load: Scan parser detected a corrupted section in TsFile %s", + source.getAbsolutePath()); + Awaitility.await() + .atMost(30, TimeUnit.SECONDS) + .untilAsserted( + () -> { + try (RandomAccessFile recentLog = new RandomAccessFile(log, "r")) { + Assert.assertTrue( + "The node log must not be truncated", recentLog.length() >= logOffset); + recentLog.seek(logOffset); + final byte[] bytes = new byte[Math.toIntExact(recentLog.length() - logOffset)]; + recentLog.readFully(bytes); + final String loadLog = new String(bytes, StandardCharsets.UTF_8); + Assert.assertTrue( + "This LOAD must reach Scan-to-Query fallback for its own source file", + loadLog.contains(fallbackMessage) + && loadLog.contains("Switch to query parsing for remaining devices.")); + } + }); + try (ResultSet schema = statement.executeQuery("show timeseries " + database + ".**")) { + Assert.assertFalse("Schema must remain absent after query fallback", schema.next()); + } + for (Map.Entry 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/pom.xml b/iotdb-core/datanode/pom.xml index f6a229e4e31e5..41df88961daed 100644 --- a/iotdb-core/datanode/pom.xml +++ b/iotdb-core/datanode/pom.xml @@ -341,6 +341,11 @@ powermock-api-mockito2 test + + org.powermock + powermock-reflect + test + org.awaitility awaitility 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 e02a1aed7b036..4c42be75b2cf1 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 @@ -64,6 +64,9 @@ private DataNodeQueryMessages() {} "There is not enough memory for Query %s, the contextHolder is %s,current remaining free " + "memory is %dB, already reserved memory for this context in total is %dB, the memory " + "requested this time is %dB"; + public static final String + EXCEPTION_THE_DATABASE_ARG_DOES_NOT_EXIST_AND_AUTO_CREATE_SCHEMA_IS_DISABLED_28F8D49A = + "The database %s does not exist and auto-create-schema is disabled."; public static final String LOG_DELETE_AFTER_LOADING_ARG_ERROR_C606A060 = "Delete After Loading {} error."; } 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 2aaee8bcfc72e..182e366139895 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 @@ -61,5 +61,8 @@ private DataNodeQueryMessages() {} public static final String QUERY_EXCEPTION_THERE_IS_NOT_ENOUGH_MEMORY_FOR_QUERY_S_THE_CONTEXTHOLDER_546CDD02 = "Query %s 内存不足，contextHolder 为 %s，当前剩余空闲内存为 %dB，该 context 已预留总内存为 %dB，本次请求内存为 %dB。"; + public static final String + EXCEPTION_THE_DATABASE_ARG_DOES_NOT_EXIST_AND_AUTO_CREATE_SCHEMA_IS_DISABLED_28F8D49A = + "数据库 %s 不存在，且 auto-create-schema 已禁用。"; public static final String LOG_DELETE_AFTER_LOADING_ARG_ERROR_C606A060 = "加载后删除 {} 时出错。"; } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/exception/LoadAnalyzeMissingDatabaseException.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/exception/LoadAnalyzeMissingDatabaseException.java new file mode 100644 index 0000000000000..b0191aad06e6c --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/exception/LoadAnalyzeMissingDatabaseException.java @@ -0,0 +1,27 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.db.exception; + +public class LoadAnalyzeMissingDatabaseException extends LoadAnalyzeMissingSchemaException { + + public LoadAnalyzeMissingDatabaseException(final String message) { + super(message); + } +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/common/SessionInfo.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/common/SessionInfo.java index dee94543cd622..718c2cd7802b3 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/common/SessionInfo.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/common/SessionInfo.java @@ -34,23 +34,33 @@ public class SessionInfo { private final String userName; private final ZoneId zoneId; private final String cliHostname; + // LOAD conversion uses this policy during local schema analysis; keep the session wire format. + private final boolean enableAutoCreateSchema; private ClientVersion version = ClientVersion.V_1_0; public SessionInfo(long sessionId, String userName, ZoneId zoneId, String cliHostname) { - this.sessionId = sessionId; - this.userName = userName; - this.zoneId = zoneId; - this.cliHostname = cliHostname; + this(sessionId, userName, zoneId, ClientVersion.V_1_0, cliHostname); } public SessionInfo( long sessionId, String userName, ZoneId zoneId, ClientVersion version, String cliHostname) { + this(sessionId, userName, zoneId, version, cliHostname, true); + } + + public SessionInfo( + long sessionId, + String userName, + ZoneId zoneId, + ClientVersion version, + String cliHostname, + boolean enableAutoCreateSchema) { this.sessionId = sessionId; this.userName = userName; this.zoneId = zoneId; this.version = version; this.cliHostname = cliHostname; + this.enableAutoCreateSchema = enableAutoCreateSchema; } public long getSessionId() { @@ -73,6 +83,10 @@ public String getCliHostname() { return cliHostname; } + public boolean isEnableAutoCreateSchema() { + return enableAutoCreateSchema; + } + public static SessionInfo deserializeFrom(ByteBuffer buffer) { long sessionId = ReadWriteIOUtils.readLong(buffer); String userName = ReadWriteIOUtils.readString(buffer); 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 7dfd99ca62886..ca9527453c82f 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 @@ -40,6 +40,7 @@ import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.exception.LoadAnalyzeException; import org.apache.iotdb.db.exception.LoadAnalyzeInvalidPathException; +import org.apache.iotdb.db.exception.LoadAnalyzeMissingDatabaseException; import org.apache.iotdb.db.exception.LoadAnalyzeMissingSchemaException; import org.apache.iotdb.db.exception.LoadAnalyzeTypeMismatchException; import org.apache.iotdb.db.exception.load.LoadEmptyFileException; @@ -191,7 +192,10 @@ public LoadTsFileAnalyzer( this.isAutoCreateSchemaRequested = loadTsFileStatement.isAutoCreateSchema(); this.isAutoCreateSchemaEnabled = IoTDBDescriptor.getInstance().getConfig().isAutoCreateSchemaEnabled() - && isAutoCreateSchemaRequested; + && isAutoCreateSchemaRequested + && (context == null + || context.getSession() == null + || context.getSession().isEnableAutoCreateSchema()); this.isAutoCreateDatabase = loadTsFileStatement.isAutoCreateDatabase(); this.isDeleteAfterLoad = loadTsFileStatement.isDeleteAfterLoad(); this.isConvertOnTypeMismatch = loadTsFileStatement.isConvertOnTypeMismatch(); @@ -275,7 +279,7 @@ private boolean doAsyncLoad(final Analysis analysis) { databaseLevel, isConvertOnTypeMismatch, isVerifySchema, - isAutoCreateSchemaRequested, + isAutoCreateSchemaEnabled, tabletConversionThresholdBytes, isGeneratedByPipe, context.getSession().getUserName()); @@ -416,10 +420,7 @@ private boolean handleSingleMiniFile(final int i) throws FileNotFoundException { final TSStatus status = loadTsFileDataTypeConverter - .convertForTreeModel( - LoadTsFileStatement.createUnchecked(tsFiles.get(i).getPath()) - .setDeleteAfterLoad(isDeleteAfterLoad) - .setConvertOnTypeMismatch(isConvertOnTypeMismatch)) + .convertForTreeModel(createTreeConversionStatement(tsFiles.get(i))) .orElse(null); if (status == null || !loadTsFileDataTypeConverter.isSuccessful(status)) { @@ -591,6 +592,19 @@ private Analysis executeTabletConversionOnException( return analysis; } + private LoadTsFileStatement createTreeConversionStatement(final File tsFile) + throws FileNotFoundException { + final LoadTsFileStatement statement = + LoadTsFileStatement.createUnchecked(tsFile.getPath()) + .setDeleteAfterLoad(isDeleteAfterLoad) + .setConvertOnTypeMismatch(isConvertOnTypeMismatch); + statement.setAutoCreateSchema(isAutoCreateSchemaRequested); + if (isGeneratedByPipe) { + statement.markIsGeneratedByPipe(); + } + return statement; + } + private boolean setTemporaryUnavailableStatusIfNecessary( final Analysis analysis, final Throwable throwable) { // Region unavailability can persist beyond recovery. Retain the file for retry; permanent @@ -649,6 +663,7 @@ boolean isTemporaryUnavailableDueToPipeSchemaNotReady(final Throwable throwable) private boolean shouldSkipConversion(LoadAnalyzeException e) { return e instanceof LoadAnalyzeInvalidPathException + || e instanceof LoadAnalyzeMissingDatabaseException || (e instanceof LoadAnalyzeTypeMismatchException) && !loadTsFileStatement.isConvertOnTypeMismatch(); } @@ -961,6 +976,14 @@ private void autoCreateDatabase() } databasesNeededToBeSet.removeAll(schemaCache.getAlreadySetDatabases()); + if (!databasesNeededToBeSet.isEmpty() && !isAutoCreateSchemaEnabled) { + throw new LoadAnalyzeMissingDatabaseException( + String.format( + DataNodeQueryMessages + .EXCEPTION_THE_DATABASE_ARG_DOES_NOT_EXIST_AND_AUTO_CREATE_SCHEMA_IS_DISABLED_28F8D49A, + databasesNeededToBeSet.iterator().next())); + } + // 2. create the databases that do not exist for (final PartialPath databasePath : databasesNeededToBeSet) { final DatabaseSchemaStatement statement = @@ -1084,7 +1107,7 @@ private ISchemaTree autoCreateSchema() throws IllegalPathException { encodingsList, compressionTypesList, isAlignedList, - isAutoCreateSchemaRequested, + isAutoCreateSchemaEnabled, context); } 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 e81abb85769e4..e45a62271334a 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 @@ -35,6 +35,8 @@ import java.util.List; import java.util.stream.Collectors; +import static org.apache.iotdb.db.queryengine.plan.analyze.schema.SchemaValidator.isAutoCreateSchemaAllowed; + class NormalSchemaFetcher { private final IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig(); @@ -168,7 +170,7 @@ List processNormalTimeSeries( } // [Step 5] Auto Create and process the missing schema - if (config.isAutoCreateSchemaEnabled()) { + if (config.isAutoCreateSchemaEnabled() && isAutoCreateSchemaAllowed(context)) { // Check the isAligned value. If the input value is different from the actual value of the // existing device, throw exception. PartialPath devicePath = schemaComputationWithAutoCreation.getDevicePath(); @@ -319,7 +321,7 @@ void processNormalTimeSeries( } // [Step 5] Auto Create and process the missing schema - if (config.isAutoCreateSchemaEnabled()) { + if (config.isAutoCreateSchemaEnabled() && isAutoCreateSchemaAllowed(context)) { List devicePathList = schemaComputationWithAutoCreationList.stream() .map(ISchemaComputationWithAutoCreation::getDevicePath) 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 443d9351cae20..cd8cc311deb7d 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 @@ -37,6 +37,12 @@ public class SchemaValidator { + static boolean isAutoCreateSchemaAllowed(final MPPQueryContext context) { + return context == null + || context.getSession() == null + || context.getSession().isEnableAutoCreateSchema(); + } + public static void validate( ISchemaFetcher schemaFetcher, InsertBaseStatement insertStatement, MPPQueryContext context) { try { @@ -86,7 +92,7 @@ public static ISchemaTree validate( final List 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 5967afceada25..3ec43270805cf 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 8784f1970813c..a2717964c6ffd 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 @@ -59,6 +59,7 @@ public class LoadSingleTsFileNode extends WritePlanNode { private final File tsFile; private final TsFileResource resource; private final boolean deleteAfterLoad; + private final boolean autoCreateSchema; private final long writePointCount; private boolean needDecodeTsFile; @@ -66,13 +67,27 @@ public class LoadSingleTsFileNode extends WritePlanNode { public LoadSingleTsFileNode( PlanNodeId id, TsFileResource resource, boolean deleteAfterLoad, long writePointCount) { + this(id, resource, deleteAfterLoad, writePointCount, true); + } + + public LoadSingleTsFileNode( + PlanNodeId id, + TsFileResource resource, + boolean deleteAfterLoad, + long writePointCount, + boolean autoCreateSchema) { super(id); this.tsFile = resource.getTsFile(); this.resource = resource; this.deleteAfterLoad = deleteAfterLoad; + this.autoCreateSchema = autoCreateSchema; this.writePointCount = writePointCount; } + public boolean isAutoCreateSchema() { + return autoCreateSchema; + } + public boolean isTsFileEmpty() { return resource.getDevices().isEmpty(); } @@ -255,11 +270,18 @@ public boolean equals(Object o) { && Objects.equals(resource, loadSingleTsFileNode.resource) && Objects.equals(needDecodeTsFile, loadSingleTsFileNode.needDecodeTsFile) && Objects.equals(deleteAfterLoad, loadSingleTsFileNode.deleteAfterLoad) + && autoCreateSchema == loadSingleTsFileNode.autoCreateSchema && Objects.equals(localRegionReplicaSet, loadSingleTsFileNode.localRegionReplicaSet); } @Override public int hashCode() { - return Objects.hash(tsFile, resource, needDecodeTsFile, deleteAfterLoad, localRegionReplicaSet); + return Objects.hash( + tsFile, + resource, + 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 540f9e4d9f6d6..f5303183c5926 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 @@ -103,7 +103,8 @@ public List splitByPartition(IAnalysis analysis) { getPlanNodeId(), resources.get(i), statement.isDeleteAfterLoad(), - statement.getWritePointCount(i))); + statement.getWritePointCount(i), + statement.isAutoCreateSchema())); } return res; } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileScheduler.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileScheduler.java index d1f1f3fb37058..28c8cc2e3cfa6 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileScheduler.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileScheduler.java @@ -569,11 +569,11 @@ private void convertFailedTsFilesToTabletsAndRetry() { final String filePath = failedNode.getTsFileResource().getTsFilePath(); try { + final LoadTsFileStatement statement = + buildRetryTreeLoadStatement(filePath, failedNode.isDeleteAfterLoad()); + statement.setAutoCreateSchema(failedNode.isAutoCreateSchema()); final TSStatus status = - loadTsFileDataTypeConverter - .convertForTreeModel( - buildRetryTreeLoadStatement(filePath, failedNode.isDeleteAfterLoad())) - .orElse(null); + loadTsFileDataTypeConverter.convertForTreeModel(statement).orElse(null); if (loadTsFileDataTypeConverter.isSuccessful(status)) { iterator.remove(); 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 d9723a223436b..df9ac0cab83f2 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 @@ -31,6 +31,7 @@ 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.SessionInfo; import org.apache.iotdb.db.queryengine.plan.Coordinator; import org.apache.iotdb.db.queryengine.plan.analyze.ClusterPartitionFetcher; import org.apache.iotdb.db.queryengine.plan.analyze.lock.DataNodeSchemaLockManager; @@ -141,10 +142,6 @@ private static class TabletConversionSemaphoreHolder { private final boolean isGeneratedByPipe; private final MPPQueryContext context; - private final LoadTreeStatementDataTypeConvertExecutionVisitor - treeStatementDataTypeConvertExecutionVisitor = - new LoadTreeStatementDataTypeConvertExecutionVisitor(this::executeForTreeModel); - public LoadTsFileDataTypeConverter( final MPPQueryContext context, final boolean isGeneratedByPipe) { this.context = context; @@ -157,7 +154,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) { @@ -177,7 +178,8 @@ public Optional convertForTreeModel(final LoadTsFileStatement loadTsFi } } - private TSStatus executeForTreeModel(final Statement statement) { + private TSStatus executeForTreeModel( + final Statement statement, final boolean allowAutoCreateSchema) { final IClientSession session; final boolean needToCreateSession = SESSION_MANAGER.getCurrSession() == null; if (needToCreateSession) { @@ -195,11 +197,23 @@ private TSStatus executeForTreeModel(final Statement statement) { } try { + final SessionInfo sessionInfo = + SESSION_MANAGER.getSessionInfo(SESSION_MANAGER.getCurrSession()); return Coordinator.getInstance() .executeForTreeModel( isGeneratedByPipe ? new PipeEnrichedStatement(statement) : statement, SESSION_MANAGER.requestQueryId(), - SESSION_MANAGER.getSessionInfo(SESSION_MANAGER.getCurrSession()), + new SessionInfo( + sessionInfo.getSessionId(), + sessionInfo.getUserName(), + sessionInfo.getZoneId(), + sessionInfo.getVersion(), + sessionInfo.getCliHostname(), + allowAutoCreateSchema + && sessionInfo.isEnableAutoCreateSchema() + && (context == null + || context.getSession() == null + || 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 6fe20c64bc2d1..48d472d70658c 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,19 +21,30 @@ import org.apache.iotdb.common.rpc.thrift.TEndPoint; import org.apache.iotdb.common.rpc.thrift.TSStatus; +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.path.PartialPath; +import org.apache.iotdb.confignode.rpc.thrift.TShowDatabaseResp; +import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.exception.LoadAnalyzeException; +import org.apache.iotdb.db.exception.LoadAnalyzeMissingDatabaseException; import org.apache.iotdb.db.exception.LoadAnalyzeTypeMismatchException; import org.apache.iotdb.db.exception.sql.SemanticException; 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.common.SessionInfo; import org.apache.iotdb.db.queryengine.plan.analyze.Analysis; import org.apache.iotdb.db.queryengine.plan.analyze.IPartitionFetcher; import org.apache.iotdb.db.queryengine.plan.analyze.schema.ISchemaFetcher; import org.apache.iotdb.db.queryengine.plan.analyze.schema.SchemaValidator; import org.apache.iotdb.db.queryengine.plan.statement.crud.LoadTsFileStatement; +import org.apache.iotdb.db.storageengine.load.active.ActiveLoadPathHelper; +import org.apache.iotdb.db.storageengine.load.active.ActiveLoadUtil; import org.apache.iotdb.db.storageengine.load.converter.LoadTsFileDataTypeConverter; import org.apache.iotdb.rpc.TSStatusCode; @@ -48,27 +59,41 @@ import org.junit.Test; import org.junit.rules.TemporaryFolder; import org.junit.runner.RunWith; +import org.mockito.ArgumentCaptor; import org.mockito.Mockito; 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 org.powermock.reflect.Whitebox; import java.io.File; import java.io.FileOutputStream; import java.lang.reflect.Field; import java.lang.reflect.Method; +import java.time.ZoneOffset; import java.util.Collections; +import java.util.HashMap; +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.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, SchemaValidator.class, ActiveLoadUtil.class}) public class LoadTsFileAnalyzerDiagnosticsTest { @Rule public TemporaryFolder temporaryFolder = new TemporaryFolder(); @@ -229,6 +254,217 @@ public void testUnavailableRegionsDuringSingleFileAnalysisShouldRetry() throws E } } + @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 requested : new boolean[] {false, true, false}) { + for (final boolean generatedByPipe : new boolean[] {false, true}) { + clearInvocations(converter); + final LoadTsFileStatement treeStatement = + LoadTsFileStatement.createUnchecked(file.getPath()); + treeStatement.setAutoCreateSchema(requested); + if (generatedByPipe) { + treeStatement.markIsGeneratedByPipe(); + } + try (final LoadTsFileAnalyzer analyzer = newAnalyzer(treeStatement, true)) { + if (miniFile) { + Assert.assertTrue((Boolean) Whitebox.invokeMethod(analyzer, "handleSingleMiniFile", 0)); + } else { + final Analysis analysis = new Analysis(); + Whitebox.invokeMethod( + analyzer, + "executeTabletConversionOnException", + 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(generatedByPipe, statement.getValue().isGeneratedByPipe()); + Assert.assertEquals(Collections.singletonList(file), statement.getValue().getTsFiles()); + Assert.assertTrue(file.exists()); + } + } + } + } + + @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, + "root.missing_db"); + try (final LoadTsFileAnalyzer analyzer = newAnalyzer(statement, true)) { + final Analysis analysis = new Analysis(); + Whitebox.invokeMethod( + analyzer, + "executeTabletConversionOnException", + 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(LoadTsFileAnalyzer.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())); + try { + IoTDBDescriptor.getInstance().getConfig().setAutoCreateSchemaEnabled(globalEnabled); + Whitebox.setInternalState( + LoadTsFileAnalyzer.class, "CONFIG_NODE_CLIENT_MANAGER", clientManager); + final File file = temporaryFolder.newFile("missing-database.tsfile"); + final LoadTsFileStatement statement = LoadTsFileStatement.createUnchecked(file.getPath()); + statement.setAutoCreateSchema(requested); + statement.setAutoCreateDatabase(true); + try (final LoadTsFileAnalyzer analyzer = newAnalyzer(statement, sessionEnabled)) { + Assert.assertFalse(analyzer.isAutoCreateSchemaEnabled()); + final Object verifier = Whitebox.getInternalState(analyzer, "schemaAutoCreatorAndVerifier"); + final Object cache = Whitebox.getInternalState(verifier, "schemaCache"); + Whitebox.invokeMethod( + cache, + "addTimeSeries", + new PlainDeviceID("root.missing_db.d"), + new MeasurementSchema("s", TSDataType.INT64)); + final LoadAnalyzeMissingDatabaseException failure = + Assert.assertThrows( + LoadAnalyzeMissingDatabaseException.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"), + failure.getMessage()); + Whitebox.invokeMethod(cache, "addAlreadySetDatabase", new PartialPath("root.missing_db")); + Whitebox.invokeMethod(verifier, "autoCreateDatabase"); + Assert.assertTrue(file.exists()); + } + verify(client, never()).setDatabase(any()); + verifyZeroInteractions(converter); + } finally { + Whitebox.setInternalState( + LoadTsFileAnalyzer.class, "CONFIG_NODE_CLIENT_MANAGER", originalClientManager); + IoTDBDescriptor.getInstance().getConfig().setAutoCreateSchemaEnabled(originalGlobalEnabled); + } + } + + @Test + public void testAsyncTreeSqlLoadPreservesSchemaCreationSetting() throws Exception { + final AtomicReference> handedOffAttributes = new AtomicReference<>(); + PowerMockito.mockStatic(ActiveLoadUtil.class); + when(ActiveLoadUtil.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(AUTO_CREATE_SCHEMA_KEY, String.valueOf(requested)); + final LoadTsFileStatement statement = + LoadTsFileStatement.createUnchecked(file.getPath()); + statement.setLoadAttributes(attributes); + try (final LoadTsFileAnalyzer analyzer = newAnalyzer(statement, sessionEnabled)) { + final Analysis analysis = analyzer.analyzeFileByFile(new Analysis()); + Assert.assertTrue(analysis.isFinishQueryAfterAnalyze()); + Assert.assertEquals( + TSStatusCode.SUCCESS_STATUS.getStatusCode(), analysis.getFailStatus().getCode()); + } + Assert.assertNotNull(handedOffAttributes.get()); + final File queuedFile = + new File( + ActiveLoadPathHelper.resolveTargetDir(pendingDir, handedOffAttributes.get()), + file.getName()); + final Map parsed = + ActiveLoadPathHelper.parseAttributes(queuedFile, pendingDir); + final LoadTsFileStatement active = LoadTsFileStatement.createUnchecked(file.getPath()); + ActiveLoadPathHelper.applyAttributesToStatement(parsed, active, false); + Assert.assertEquals( + globalEnabled && requested && sessionEnabled, active.isAutoCreateSchema()); + Assert.assertEquals(requested, statement.isAutoCreateSchema()); + Assert.assertTrue(file.exists()); + } + } + } + } finally { + IoTDBDescriptor.getInstance().getConfig().setAutoCreateSchemaEnabled(originalGlobalEnabled); + } + } + + private LoadTsFileAnalyzer newAnalyzer( + final LoadTsFileStatement statement, final boolean allowed) { + return new LoadTsFileAnalyzer( + statement, + new MPPQueryContext( + "", + new QueryId("tree_schema_policy"), + new SessionInfo( + 0, "root", ZoneOffset.UTC, IoTDBConstant.ClientVersion.V_1_0, "", allowed), + null, + null), + mock(IPartitionFetcher.class), + mock(ISchemaFetcher.class)); + } + private TSStatus analyzeAfterSchemaFailure(final TSStatusCode code) throws Exception { return analyzeAfterSchemaFailure( new IoTDBRuntimeException("schema request failed", code.getStatusCode())); 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 0f3e2b1913458..adc7c696d32a2 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,12 +19,16 @@ package org.apache.iotdb.db.queryengine.plan.analyze.load; +import org.apache.iotdb.commons.conf.IoTDBConstant; import org.apache.iotdb.commons.path.PartialPath; import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.exception.LoadAnalyzeInvalidPathException; import org.apache.iotdb.db.exception.LoadAnalyzeMissingSchemaException; import org.apache.iotdb.db.queryengine.common.MPPQueryContext; import org.apache.iotdb.db.queryengine.common.QueryId; +import org.apache.iotdb.db.queryengine.common.SessionInfo; +import org.apache.iotdb.db.queryengine.common.schematree.ClusterSchemaTree; +import org.apache.iotdb.db.queryengine.plan.analyze.Analysis; import org.apache.iotdb.db.queryengine.plan.analyze.IPartitionFetcher; import org.apache.iotdb.db.queryengine.plan.analyze.schema.ISchemaFetcher; import org.apache.iotdb.db.queryengine.plan.statement.crud.LoadTsFileStatement; @@ -32,6 +36,9 @@ import org.apache.tsfile.enums.TSDataType; import org.apache.tsfile.file.metadata.IDeviceID; import org.apache.tsfile.file.metadata.PlainDeviceID; +import org.apache.tsfile.read.common.Path; +import org.apache.tsfile.write.TsFileWriter; +import org.apache.tsfile.write.record.Tablet; import org.apache.tsfile.write.schema.MeasurementSchema; import org.junit.Assert; import org.junit.Test; @@ -40,12 +47,19 @@ import java.lang.reflect.Field; import java.lang.reflect.InvocationTargetException; import java.lang.reflect.Method; +import java.time.ZoneOffset; import java.util.Arrays; import java.util.Collections; import java.util.HashSet; +import java.util.List; import java.util.Set; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyList; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; public class LoadTsFileAnalyzerTest { @@ -169,6 +183,70 @@ public void testGlobalAutoCreateDisabledKeepsPerLoadAutoCreatePermission() throw } } + @Test + public void testTreeTsFileAnalysisPreservesSessionAutoCreateSchemaSetting() throws Exception { + final boolean originalEnabled = + IoTDBDescriptor.getInstance().getConfig().isAutoCreateSchemaEnabled(); + final long originalThreshold = + 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 (final 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); + tablet.rowSize = 1; + writer.write(tablet); + } + try { + IoTDBDescriptor.getInstance().getConfig().setAutoCreateSchemaEnabled(true); + IoTDBDescriptor.getInstance().getConfig().setLoadTabletConversionThresholdBytes(-1); + final SessionInfo session = + new SessionInfo(0, "root", ZoneOffset.UTC, IoTDBConstant.ClientVersion.V_1_0, "", false); + final MPPQueryContext context = + new MPPQueryContext("", new QueryId("tree_load_session_schema"), session, null, null); + final LoadTsFileStatement statement = LoadTsFileStatement.createUnchecked(tsFile.getPath()); + statement.setAutoCreateSchema(true); + statement.setAutoCreateDatabase(false); + statement.setVerifySchema(true); + final ISchemaFetcher fetcher = mock(ISchemaFetcher.class); + final ClusterSchemaTree existing = new ClusterSchemaTree(); + existing.appendSingleMeasurement( + new PartialPath("root.load_session_schema.d1.s1"), + schemas.get(0), + null, + null, + null, + false); + when(fetcher.fetchSchemaList(anyList(), anyList(), any(MPPQueryContext.class))) + .thenReturn(existing); + try (final LoadTsFileAnalyzer analyzer = + new LoadTsFileAnalyzer(statement, context, mock(IPartitionFetcher.class), fetcher)) { + Assert.assertFalse(analyzer.analyzeFileByFile(new Analysis()).isFailed()); + } + verify(fetcher).fetchSchemaList(anyList(), anyList(), any(MPPQueryContext.class)); + verify(fetcher, never()) + .fetchSchemaListWithAutoCreate( + anyList(), + anyList(), + anyList(), + anyList(), + anyList(), + anyList(), + any(MPPQueryContext.class)); + Assert.assertSame(session, context.getSession()); + Assert.assertFalse(context.getSession().isEnableAutoCreateSchema()); + } finally { + IoTDBDescriptor.getInstance().getConfig().setAutoCreateSchemaEnabled(originalEnabled); + IoTDBDescriptor.getInstance() + .getConfig() + .setLoadTabletConversionThresholdBytes(originalThreshold); + Assert.assertTrue(tsFile.delete()); + } + } + private LoadTsFileAnalyzer createAnalyzer(final File tsFile) throws Exception { return createAnalyzer( LoadTsFileStatement.createUnchecked(tsFile.getAbsolutePath()), "load_tree_test"); 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..ad09e5d55d8cd --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/analyze/schema/TreeSchemaAutoCreationPolicyTest.java @@ -0,0 +1,200 @@ +/* + * 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.conf.IoTDBConstant; +import org.apache.iotdb.commons.path.MeasurementPath; +import org.apache.iotdb.commons.path.PartialPath; +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.SessionInfo; +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.analyze.cache.schema.DataNodeSchemaCache; +import org.apache.iotdb.db.schemaengine.template.Template; + +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 DataNodeSchemaCache 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(DataNodeSchemaCache.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) { + return new MPPQueryContext( + "", + new QueryId("load_policy"), + new SessionInfo(0, "root", ZoneOffset.UTC, IoTDBConstant.ClientVersion.V_1_0, "", allowed), + null, + null); + } +} diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadFallbackSchemaPolicyTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadFallbackSchemaPolicyTest.java new file mode 100644 index 0000000000000..a5f773fad546f --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadFallbackSchemaPolicyTest.java @@ -0,0 +1,135 @@ +/* + * 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.scheduler.load; + +import org.apache.iotdb.common.rpc.thrift.TSStatus; +import org.apache.iotdb.commons.client.IClientManager; +import org.apache.iotdb.db.queryengine.common.MPPQueryContext; +import org.apache.iotdb.db.queryengine.common.PlanFragmentId; +import org.apache.iotdb.db.queryengine.execution.QueryStateMachine; +import org.apache.iotdb.db.queryengine.plan.analyze.Analysis; +import org.apache.iotdb.db.queryengine.plan.analyze.IPartitionFetcher; +import org.apache.iotdb.db.queryengine.plan.planner.plan.DistributedQueryPlan; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeId; +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.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.memory.LoadTsFileMemoryManager; +import org.apache.iotdb.rpc.TSStatusCode; + +import org.junit.Assert; +import org.junit.Rule; +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; +import org.powermock.modules.junit4.PowerMockRunner; +import org.powermock.reflect.Whitebox; + +import java.io.File; +import java.util.Collections; +import java.util.List; +import java.util.Optional; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.RETURNS_DEEP_STUBS; +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) +@PowerMockIgnore({"com.sun.org.apache.xerces.*", "javax.xml.*", "org.xml.*", "javax.management.*"}) +@PrepareForTest(LoadTsFileScheduler.class) +public class LoadFallbackSchemaPolicyTest { + @Rule public final TemporaryFolder temporaryFolder = new TemporaryFolder(); + + @Test + public void testTreeFallbackPreservesRequestSchemaCreationSetting() throws Exception { + final LoadTsFileDataTypeConverter converter = mock(LoadTsFileDataTypeConverter.class); + when(converter.convertForTreeModel(any())) + .thenReturn(Optional.of(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()))); + when(converter.isSuccessful(any())).thenCallRealMethod(); + PowerMockito.whenNew(LoadTsFileDataTypeConverter.class) + .withAnyArguments() + .thenReturn(converter); + final File file = temporaryFolder.newFile("tree-schema.tsfile"); + for (final boolean generatedByPipe : new boolean[] {false, true}) { + for (final boolean requested : new boolean[] {false, true, false}) { + clearInvocations(converter); + final LoadTsFileStatement original = LoadTsFileStatement.createUnchecked(file.getPath()); + original.setAutoCreateSchema(requested); + original.setDeleteAfterLoad(true); + original.addWritePointCount(1); + final Analysis analysis = new Analysis(); + analysis.setStatement(generatedByPipe ? new PipeEnrichedStatement(original) : original); + final LoadSingleTsFileNode node = + (LoadSingleTsFileNode) + new LoadTsFileNode( + new PlanNodeId("load_schema"), + Collections.singletonList(new TsFileResource(file))) + .splitByPartition(analysis) + .get(0); + final DistributedQueryPlan plan = mock(DistributedQueryPlan.class, RETURNS_DEEP_STUBS); + when(plan.getRootSubPlan().getPlanFragment().getId()) + .thenReturn(new PlanFragmentId("test", 0)); + when(plan.getInstances()).thenReturn(Collections.emptyList()); + final QueryStateMachine state = mock(QueryStateMachine.class); + final LoadTsFileScheduler scheduler = + new LoadTsFileScheduler( + plan, + mock(MPPQueryContext.class), + state, + mock(IClientManager.class), + mock(IPartitionFetcher.class), + generatedByPipe, + "root.test"); + try { + final List nodes = + Whitebox.getInternalState(scheduler, "tsFileNodeList"); + final List failed = + Whitebox.getInternalState(scheduler, "failedTsFileNodeIndexes"); + nodes.add(node); + failed.add(0); + Whitebox.invokeMethod(scheduler, "convertFailedTsFilesToTabletsAndRetry"); + verify(state).transitionToFinished(); + Assert.assertTrue(failed.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.assertEquals("root.test", statement.getValue().getDatabase()); + Assert.assertTrue(file.exists()); + } finally { + LoadTsFileMemoryManager.getInstance().releaseDataCacheMemoryBlock(); + } + } + } + } +} 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..a2039c0c4f193 --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/converter/LoadTsFileDataTypeConverterSchemaTest.java @@ -0,0 +1,197 @@ +/* + * 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.conf.IoTDBConstant; +import org.apache.iotdb.db.conf.IoTDBDescriptor; +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.common.SessionInfo; +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.statement.Statement; +import org.apache.iotdb.db.queryengine.plan.statement.crud.LoadTsFileStatement; +import org.apache.iotdb.rpc.TSStatusCode; + +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.read.common.Path; +import org.apache.tsfile.write.TsFileWriter; +import org.apache.tsfile.write.record.Tablet; +import org.apache.tsfile.write.schema.MeasurementSchema; +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.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) +public class LoadTsFileDataTypeConverterSchemaTest { + @Rule public final TemporaryFolder temporaryFolder = new TemporaryFolder(); + private final List conversionSessions = new ArrayList<>(); + private InternalClientSession session; + private int originalDataNodeId; + 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.setUsername("root"); + session.setClientVersion(IoTDBConstant.ClientVersion.V_1_0); + session.setZoneId(ZoneOffset.UTC); + SessionManager.getInstance().registerSession(session); + PowerMockito.mockStatic(Coordinator.class); + final Coordinator coordinator = mock(Coordinator.class); + when(Coordinator.getInstance()).thenReturn(coordinator); + when(coordinator.executeForTreeModel( + any(Statement.class), + anyLong(), + any(SessionInfo.class), + anyString(), + any(IPartitionFetcher.class), + any(ISchemaFetcher.class), + anyLong(), + anyBoolean(), + anyBoolean())) + .thenAnswer( + invocation -> { + conversionSessions.add(invocation.getArgument(2)); + return new ExecutionResult( + new QueryId("tree_load_schema_conversion"), + new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode())); + }); + treeTsFile = temporaryFolder.newFile("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); + tablet.rowSize = 1; + writer.write(tablet); + } + } + + @After + public void tearDown() { + SessionManager.getInstance().removeCurrSession(); + IoTDBDescriptor.getInstance().getConfig().setDataNodeId(originalDataNodeId); + } + + @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.assertSame(session, SessionManager.getInstance().getCurrSession()); + } + } finally { + DataNodeSchemaLockManager.getInstance().releaseReadLock(context); + } + } + } + + @Test + public void testTreeConversionPreservesDisabledQuerySession() throws Exception { + assertDisabledQuerySession(false); + } + + @Test + public void testTreeConversionPreservesDisabledQuerySessionWithoutClientSession() + throws Exception { + assertDisabledQuerySession(true); + } + + private void assertDisabledQuerySession(final boolean withoutClientSession) throws Exception { + if (withoutClientSession) { + SessionManager.getInstance().removeCurrSession(); + } + final SessionInfo querySession = + new SessionInfo( + session.getId(), "root", ZoneOffset.UTC, IoTDBConstant.ClientVersion.V_1_0, "", false); + final MPPQueryContext context = + new MPPQueryContext("", new QueryId("tree_load_schema"), querySession, null, null); + try { + assertTreeConversionSchemaFlag(new LoadTsFileDataTypeConverter(context, false), true, false); + Assert.assertSame(querySession, context.getSession()); + Assert.assertFalse(querySession.isEnableAutoCreateSchema()); + if (withoutClientSession) { + Assert.assertNull(SessionManager.getInstance().getCurrSession()); + } else { + Assert.assertSame(session, SessionManager.getInstance().getCurrSession()); + } + } finally { + DataNodeSchemaLockManager.getInstance().releaseReadLock(context); + } + } + + private void assertTreeConversionSchemaFlag( + final LoadTsFileDataTypeConverter converter, final boolean requested, final boolean expected) + throws Exception { + conversionSessions.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(conversionSessions.isEmpty()); + for (final SessionInfo conversionSession : conversionSessions) { + Assert.assertEquals(expected, conversionSession.isEnableAutoCreateSchema()); + Assert.assertEquals("root", conversionSession.getUserName()); + } + Assert.assertEquals(requested, statement.isAutoCreateSchema()); + Assert.assertTrue(treeTsFile.exists()); + } +}