From 4cc75a8e74fc83910c74dc37e9ed2364dce31f9e Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Mon, 27 Jul 2026 15:09:35 +0800 Subject: [PATCH 1/2] [Pipe] Fix historical TsFile schema recovery race --- .../enhanced/IoTDBPipeAutoConflictIT.java | 5 ++ .../task/builder/PipeDataNodeTaskBuilder.java | 56 +++++++++++-- .../thrift/IoTDBDataNodeReceiver.java | 76 +++++++++++++----- .../PipeTransferTsFileSealWithModReq.java | 72 +++++++++++++++-- .../airgap/IoTDBDataRegionAirGapSink.java | 9 ++- .../handler/PipeTransferTsFileHandler.java | 8 +- .../thrift/sync/IoTDBDataRegionSyncSink.java | 8 +- .../plan/analyze/load/LoadTsFileAnalyzer.java | 26 ++++-- .../TreeSchemaAutoCreatorAndVerifier.java | 1 + .../plan/analyze/schema/ISchemaFetcher.java | 17 ++++ .../plan/analyze/schema/SchemaValidator.java | 34 +++++++- .../plan/relational/sql/ast/LoadTsFile.java | 9 +++ .../statement/crud/LoadTsFileStatement.java | 18 +++++ .../load/active/ActiveLoadPathHelper.java | 10 +++ .../load/config/LoadTsFileConfigurator.java | 23 ++++++ .../builder/PipeDataNodeTaskBuilderTest.java | 79 +++++++++++++++++++ .../thrift/IoTDBDataNodeReceiverTest.java | 30 +++++++ .../sink/PipeDataNodeThriftRequestTest.java | 13 +++ .../analyze/load/LoadTsFileAnalyzerTest.java | 61 ++++++++++++++ .../load/active/ActiveLoadDirScannerTest.java | 2 +- .../load/active/ActiveLoadPathHelperTest.java | 29 ++++++- .../pipe/config/constant/SystemConstant.java | 3 + .../options/PipeInclusionOptions.java | 15 ++++ .../commons/pipe/sink/protocol/IoTDBSink.java | 8 ++ 24 files changed, 564 insertions(+), 48 deletions(-) diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/enhanced/IoTDBPipeAutoConflictIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/enhanced/IoTDBPipeAutoConflictIT.java index 6953a5a23f4a9..8562194be996e 100644 --- a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/enhanced/IoTDBPipeAutoConflictIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/enhanced/IoTDBPipeAutoConflictIT.java @@ -438,6 +438,11 @@ public void testHistoricalActivationRace() throws Exception { Assert.assertEquals( TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("testPipe").getCode()); + TestUtils.assertDataEventuallyOnEnv( + receiverEnv, + "show paths set device template aligned_template", + "Paths,", + Collections.singleton("root.sg_aligned.device_aligned,")); TestUtils.assertDataEventuallyOnEnv( receiverEnv, "count devices root.sg_aligned.**", diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/builder/PipeDataNodeTaskBuilder.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/builder/PipeDataNodeTaskBuilder.java index 98157a2a49a09..ae4ef473998d0 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/builder/PipeDataNodeTaskBuilder.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/builder/PipeDataNodeTaskBuilder.java @@ -57,11 +57,20 @@ import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_REALTIME_ENABLE_DEFAULT_VALUE; import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_REALTIME_ENABLE_KEY; import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_REALTIME_ENABLE_KEY; +import static org.apache.iotdb.commons.pipe.datastructure.options.PipeInclusionOptions.areOptionsEnabled; public class PipeDataNodeTaskBuilder { private static final Logger LOGGER = LoggerFactory.getLogger(PipeDataNodeTaskBuilder.class); + private static final String[] SCHEMA_OPTIONS_REQUIRED_BEFORE_LOAD = { + "schema.timeseries.ordinary.create", + "schema.timeseries.template.create", + "schema.timeseries.template.alter", + "schema.timeseries.template.set", + "schema.timeseries.template.activate" + }; + private final PipeStaticMeta pipeStaticMeta; private final int regionId; private final PipeTaskMeta pipeTaskMeta; @@ -265,13 +274,46 @@ private static void checkConflict( private static void injectParameters( final PipeParameters sourceParameters, final PipeParameters sinkParameters) { - final boolean isSourceExternal = - !BuiltinPipePlugin.BUILTIN_SOURCES.contains( - sourceParameters - .getStringOrDefault( - Arrays.asList(PipeSourceConstant.EXTRACTOR_KEY, PipeSourceConstant.SOURCE_KEY), - BuiltinPipePlugin.IOTDB_EXTRACTOR.getPipePluginName()) - .toLowerCase()); + final String sourcePluginName = + sourceParameters + .getStringOrDefault( + Arrays.asList(PipeSourceConstant.EXTRACTOR_KEY, PipeSourceConstant.SOURCE_KEY), + BuiltinPipePlugin.IOTDB_EXTRACTOR.getPipePluginName()) + .toLowerCase(); + final boolean isIoTDBSource = + BuiltinPipePlugin.IOTDB_EXTRACTOR.getPipePluginName().equals(sourcePluginName) + || BuiltinPipePlugin.IOTDB_SOURCE.getPipePluginName().equals(sourcePluginName); + final boolean shouldMarkAsGeneralWriteRequest = + sinkParameters.getBooleanOrDefault( + Arrays.asList( + PipeSinkConstant.CONNECTOR_MARK_AS_GENERAL_WRITE_REQUEST_KEY, + PipeSinkConstant.SINK_MARK_AS_GENERAL_WRITE_REQUEST_KEY), + PipeSinkConstant.CONNECTOR_MARK_AS_GENERAL_WRITE_REQUEST_DEFAULT_VALUE); + final boolean shouldMarkAsPipeRequest = + !shouldMarkAsGeneralWriteRequest + && sinkParameters.getBooleanOrDefault( + Arrays.asList( + PipeSinkConstant.CONNECTOR_MARK_AS_PIPE_REQUEST_KEY, + PipeSinkConstant.SINK_MARK_AS_PIPE_REQUEST_KEY), + PipeSinkConstant.CONNECTOR_MARK_AS_PIPE_REQUEST_DEFAULT_VALUE); + + boolean shouldWaitForSchemaBeforeLoad = false; + try { + shouldWaitForSchemaBeforeLoad = + isIoTDBSource + && shouldMarkAsPipeRequest + && areOptionsEnabled(sourceParameters, SCHEMA_OPTIONS_REQUIRED_BEFORE_LOAD); + } catch (final IllegalPathException e) { + LOGGER.warn( + DataNodePipeMessages.PIPEDATANODETASKBUILDER_FAILED_TO_PARSE_INCLUSION_AND_EXCLUSION, + e.getMessage(), + e); + } + sinkParameters.addAttribute( + SystemConstant.SINK_WAIT_FOR_SCHEMA_BEFORE_LOAD_KEY, + Boolean.toString(shouldWaitForSchemaBeforeLoad)); + + final boolean isSourceExternal = !BuiltinPipePlugin.BUILTIN_SOURCES.contains(sourcePluginName); final String sinkPluginName = sinkParameters diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java index 4cca55d705f63..5d45383998cfa 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java @@ -610,34 +610,42 @@ protected String getSenderPort() { protected TSStatus loadFileV1(final PipeTransferFileSealReqV1 req, final String fileAbsolutePath) throws IOException { return isUsingAsyncLoadTsFileStrategy.get() - ? loadTsFileAsync(null, Collections.singletonList(fileAbsolutePath)) - : loadTsFileSync(null, fileAbsolutePath); + ? loadTsFileAsync(null, Collections.singletonList(fileAbsolutePath), false) + : loadTsFileSync(null, fileAbsolutePath, false); } @Override protected TSStatus loadFileV2( final PipeTransferFileSealReqV2 req, final List fileAbsolutePaths) throws IOException, IllegalPathException { - return req instanceof PipeTransferTsFileSealWithModReq - // TsFile's absolute path will be the second element - ? (isUsingAsyncLoadTsFileStrategy.get() - ? loadTsFileAsync( - ((PipeTransferTsFileSealWithModReq) req).getDatabaseNameByTsFileName(), - fileAbsolutePaths) - : loadTsFileSync( - ((PipeTransferTsFileSealWithModReq) req).getDatabaseNameByTsFileName(), - fileAbsolutePaths.get(req.getFileNames().size() - 1))) - : loadSchemaSnapShot(req.getParameters(), fileAbsolutePaths); + if (!(req instanceof PipeTransferTsFileSealWithModReq)) { + return loadSchemaSnapShot(req.getParameters(), fileAbsolutePaths); + } + + final PipeTransferTsFileSealWithModReq tsFileSealReq = (PipeTransferTsFileSealWithModReq) req; + final String databaseName = tsFileSealReq.getDatabaseNameByTsFileName(); + final boolean shouldWaitForSchemaBeforeLoad = tsFileSealReq.shouldWaitForSchemaBeforeLoad(); + // TsFile's absolute path will be the second element when the request contains a mod file. + return isUsingAsyncLoadTsFileStrategy.get() + ? loadTsFileAsync(databaseName, fileAbsolutePaths, shouldWaitForSchemaBeforeLoad) + : loadTsFileSync( + databaseName, + fileAbsolutePaths.get(req.getFileNames().size() - 1), + shouldWaitForSchemaBeforeLoad); } - private TSStatus loadTsFileAsync(final String dataBaseName, final List absolutePaths) + private TSStatus loadTsFileAsync( + final String dataBaseName, + final List absolutePaths, + final boolean shouldWaitForSchemaBeforeLoad) throws IOException { final Map loadAttributes = buildLoadTsFileAttributesForAsync( dataBaseName, shouldConvertDataTypeOnTypeMismatch, validateTsFile.get(), - shouldMarkAsPipeRequest.get()); + shouldMarkAsPipeRequest.get(), + shouldWaitForSchemaBeforeLoad); if (!LoadUtil.loadFilesToActiveDir(loadAttributes, absolutePaths, true)) { throw new PipeException(DataNodePipeMessages.LOAD_ACTIVE_LISTENING_PIPE_DIR_IS_NOT); @@ -650,24 +658,43 @@ static Map buildLoadTsFileAttributesForAsync( final boolean shouldConvertDataTypeOnTypeMismatch, final boolean validateTsFile, final boolean shouldMarkAsPipeRequest) { + return buildLoadTsFileAttributesForAsync( + dataBaseName, + shouldConvertDataTypeOnTypeMismatch, + validateTsFile, + shouldMarkAsPipeRequest, + false); + } + + static Map buildLoadTsFileAttributesForAsync( + final String dataBaseName, + final boolean shouldConvertDataTypeOnTypeMismatch, + final boolean validateTsFile, + final boolean shouldMarkAsPipeRequest, + final boolean shouldWaitForSchemaBeforeLoad) { return ActiveLoadPathHelper.buildAttributes( dataBaseName, LoadTsFileStatement.getDatabaseLevelByTreeDatabase(dataBaseName), shouldConvertDataTypeOnTypeMismatch, - validateTsFile || shouldConvertDataTypeOnTypeMismatch, + validateTsFile || shouldConvertDataTypeOnTypeMismatch || shouldWaitForSchemaBeforeLoad, + !shouldWaitForSchemaBeforeLoad, null, shouldMarkAsPipeRequest, AuthorityChecker.SUPER_USER); } - private TSStatus loadTsFileSync(final String dataBaseName, final String fileAbsolutePath) + private TSStatus loadTsFileSync( + final String dataBaseName, + final String fileAbsolutePath, + final boolean shouldWaitForSchemaBeforeLoad) throws FileNotFoundException { return executeStatementAndClassifyExceptions( buildLoadTsFileStatementForSync( dataBaseName, fileAbsolutePath, validateTsFile.get(), - shouldConvertDataTypeOnTypeMismatch)); + shouldConvertDataTypeOnTypeMismatch, + shouldWaitForSchemaBeforeLoad)); } static LoadTsFileStatement buildLoadTsFileStatementForSync( @@ -676,10 +703,23 @@ static LoadTsFileStatement buildLoadTsFileStatementForSync( final boolean validateTsFile, final boolean shouldConvertDataTypeOnTypeMismatch) throws FileNotFoundException { + return buildLoadTsFileStatementForSync( + dataBaseName, fileAbsolutePath, validateTsFile, shouldConvertDataTypeOnTypeMismatch, false); + } + + static LoadTsFileStatement buildLoadTsFileStatementForSync( + final String dataBaseName, + final String fileAbsolutePath, + final boolean validateTsFile, + final boolean shouldConvertDataTypeOnTypeMismatch, + final boolean shouldWaitForSchemaBeforeLoad) + throws FileNotFoundException { final LoadTsFileStatement statement = LoadTsFileStatement.createUnchecked(fileAbsolutePath); statement.setDeleteAfterLoad(true); statement.setConvertOnTypeMismatch(shouldConvertDataTypeOnTypeMismatch); - statement.setVerifySchema(validateTsFile || shouldConvertDataTypeOnTypeMismatch); + statement.setVerifySchema( + validateTsFile || shouldConvertDataTypeOnTypeMismatch || shouldWaitForSchemaBeforeLoad); + statement.setAutoCreateSchema(!shouldWaitForSchemaBeforeLoad); statement.setAutoCreateDatabase( IoTDBDescriptor.getInstance().getConfig().isAutoCreateSchemaEnabled()); statement.setDatabase(dataBaseName); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/request/PipeTransferTsFileSealWithModReq.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/request/PipeTransferTsFileSealWithModReq.java index 25fb874cb6ea6..70a965729950e 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/request/PipeTransferTsFileSealWithModReq.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/request/PipeTransferTsFileSealWithModReq.java @@ -40,6 +40,7 @@ protected PipeRequestType getPlanType() { } protected static final String DATABASE_NAME_KEY_PREFIX = "DATABASE_NAME_"; + private static final String WAIT_FOR_SCHEMA_BEFORE_LOAD_KEY = "WAIT_FOR_SCHEMA_BEFORE_LOAD"; public String getDatabaseNameByTsFileName() { return parameters == null @@ -47,6 +48,11 @@ public String getDatabaseNameByTsFileName() { : parameters.get(generateDatabaseNameWithFileNameKey(fileNames.get(fileNames.size() - 1))); } + public boolean shouldWaitForSchemaBeforeLoad() { + return parameters != null + && Boolean.parseBoolean(parameters.get(WAIT_FOR_SCHEMA_BEFORE_LOAD_KEY)); + } + protected static String generateDatabaseNameWithFileNameKey(final String fileName) { return DATABASE_NAME_KEY_PREFIX + fileName; } @@ -69,25 +75,44 @@ public static PipeTransferTsFileSealWithModReq toTPipeTransferReq( final long tsFileLength, final String dataBaseName) throws IOException { + return toTPipeTransferReq( + modFileName, modFileLength, tsFileName, tsFileLength, dataBaseName, false); + } + + public static PipeTransferTsFileSealWithModReq toTPipeTransferReq( + final String modFileName, + final long modFileLength, + final String tsFileName, + final long tsFileLength, + final String dataBaseName, + final boolean shouldWaitForSchemaBeforeLoad) + throws IOException { return (PipeTransferTsFileSealWithModReq) new PipeTransferTsFileSealWithModReq() .convertToTPipeTransferReq( Arrays.asList(modFileName, tsFileName), Arrays.asList(modFileLength, tsFileLength), - Collections.singletonMap( - generateDatabaseNameWithFileNameKey(tsFileName), dataBaseName)); + generateParameters(tsFileName, dataBaseName, shouldWaitForSchemaBeforeLoad)); } public static PipeTransferTsFileSealWithModReq toTPipeTransferReq( final String tsFileName, final long tsFileLength, final String dataBaseName) throws IOException { + return toTPipeTransferReq(tsFileName, tsFileLength, dataBaseName, false); + } + + public static PipeTransferTsFileSealWithModReq toTPipeTransferReq( + final String tsFileName, + final long tsFileLength, + final String dataBaseName, + final boolean shouldWaitForSchemaBeforeLoad) + throws IOException { return (PipeTransferTsFileSealWithModReq) new PipeTransferTsFileSealWithModReq() .convertToTPipeTransferReq( Collections.singletonList(tsFileName), Collections.singletonList(tsFileLength), - Collections.singletonMap( - generateDatabaseNameWithFileNameKey(tsFileName), dataBaseName)); + generateParameters(tsFileName, dataBaseName, shouldWaitForSchemaBeforeLoad)); } public static PipeTransferTsFileSealWithModReq fromTPipeTransferReq(final TPipeTransferReq req) { @@ -117,23 +142,54 @@ public static byte[] toTPipeTransferBytes( final long tsFileLength, final String dataBaseName) throws IOException { + return toTPipeTransferBytes( + modFileName, modFileLength, tsFileName, tsFileLength, dataBaseName, false); + } + + public static byte[] toTPipeTransferBytes( + final String modFileName, + final long modFileLength, + final String tsFileName, + final long tsFileLength, + final String dataBaseName, + final boolean shouldWaitForSchemaBeforeLoad) + throws IOException { return new PipeTransferTsFileSealWithModReq() .convertToTPipeTransferSnapshotSealBytes( Arrays.asList(modFileName, tsFileName), Arrays.asList(modFileLength, tsFileLength), - Collections.singletonMap( - generateDatabaseNameWithFileNameKey(tsFileName), dataBaseName)); + generateParameters(tsFileName, dataBaseName, shouldWaitForSchemaBeforeLoad)); } public static byte[] toTPipeTransferBytes( final String tsFileName, final long tsFileLength, final String dataBaseName) throws IOException { + return toTPipeTransferBytes(tsFileName, tsFileLength, dataBaseName, false); + } + + public static byte[] toTPipeTransferBytes( + final String tsFileName, + final long tsFileLength, + final String dataBaseName, + final boolean shouldWaitForSchemaBeforeLoad) + throws IOException { return new PipeTransferTsFileSealWithModReq() .convertToTPipeTransferSnapshotSealBytes( Collections.singletonList(tsFileName), Collections.singletonList(tsFileLength), - Collections.singletonMap( - generateDatabaseNameWithFileNameKey(tsFileName), dataBaseName)); + generateParameters(tsFileName, dataBaseName, shouldWaitForSchemaBeforeLoad)); + } + + private static HashMap generateParameters( + final String tsFileName, + final String dataBaseName, + final boolean shouldWaitForSchemaBeforeLoad) { + final HashMap parameters = new HashMap<>(); + parameters.put(generateDatabaseNameWithFileNameKey(tsFileName), dataBaseName); + if (shouldWaitForSchemaBeforeLoad) { + parameters.put(WAIT_FOR_SCHEMA_BEFORE_LOAD_KEY, Boolean.TRUE.toString()); + } + return parameters; } /////////////////////////////// Object /////////////////////////////// diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/airgap/IoTDBDataRegionAirGapSink.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/airgap/IoTDBDataRegionAirGapSink.java index a535f5bb33a30..0cd7115bc3779 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/airgap/IoTDBDataRegionAirGapSink.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/airgap/IoTDBDataRegionAirGapSink.java @@ -475,7 +475,12 @@ private void doTransfer( if (!sendWeighted( socket, PipeTransferTsFileSealWithModReq.toTPipeTransferBytes( - modFile.getName(), modFile.length(), tsFile.getName(), tsFile.length(), dataBaseName), + modFile.getName(), + modFile.length(), + tsFile.getName(), + tsFile.length(), + dataBaseName, + shouldWaitForSchemaBeforeLoad), pipe2WeightMap)) { receiverStatusHandler.handle( new TSStatus(TSStatusCode.PIPE_RECEIVER_USER_CONFLICT_EXCEPTION.getStatusCode()) @@ -490,7 +495,7 @@ private void doTransfer( if (!sendWeighted( socket, PipeTransferTsFileSealWithModReq.toTPipeTransferBytes( - tsFile.getName(), tsFile.length(), dataBaseName), + tsFile.getName(), tsFile.length(), dataBaseName, shouldWaitForSchemaBeforeLoad), pipe2WeightMap)) { receiverStatusHandler.handle( new TSStatus(TSStatusCode.PIPE_RECEIVER_USER_CONFLICT_EXCEPTION.getStatusCode()) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTsFileHandler.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTsFileHandler.java index 7f06b771a0cb4..89791928a01d4 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTsFileHandler.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTsFileHandler.java @@ -206,9 +206,13 @@ public void transfer( modFile.length(), tsFile.getName(), tsFile.length(), - dataBaseName) + dataBaseName, + sink.shouldWaitForSchemaBeforeLoad()) : PipeTransferTsFileSealWithModReq.toTPipeTransferReq( - tsFile.getName(), tsFile.length(), dataBaseName); + tsFile.getName(), + tsFile.length(), + dataBaseName, + sink.shouldWaitForSchemaBeforeLoad()); final TPipeTransferReq req = sink.compressIfNeeded(uncompressedReq); pipeName2WeightMap.forEach( diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/sync/IoTDBDataRegionSyncSink.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/sync/IoTDBDataRegionSyncSink.java index 46d4034cf35f6..25a4f11f16e3d 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/sync/IoTDBDataRegionSyncSink.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/sync/IoTDBDataRegionSyncSink.java @@ -553,7 +553,8 @@ private void doTransfer( modFile.length(), tsFile.getName(), tsFile.length(), - dataBaseName)); + dataBaseName, + shouldWaitForSchemaBeforeLoad)); pipeName2WeightMap.forEach( (pipePair, weight) -> @@ -583,7 +584,10 @@ private void doTransfer( final TPipeTransferReq req = compressIfNeeded( PipeTransferTsFileSealWithModReq.toTPipeTransferReq( - tsFile.getName(), tsFile.length(), dataBaseName)); + tsFile.getName(), + tsFile.length(), + dataBaseName, + shouldWaitForSchemaBeforeLoad)); pipeName2WeightMap.forEach( (pipePair, weight) -> 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 48d391658403a..aa274a100c049 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 @@ -114,6 +114,8 @@ public class LoadTsFileAnalyzer implements AutoCloseable { private String databaseForTableData; private final boolean isAsyncLoad; private final boolean isVerifySchema; + private final boolean isAutoCreateSchemaAllowed; + private final boolean isAutoCreateSchema; private final boolean isAutoCreateDatabase; private final boolean isDeleteAfterLoad; private final boolean isConvertOnTypeMismatch; @@ -141,6 +143,10 @@ public LoadTsFileAnalyzer( this.databaseForTableData = loadTsFileStatement.getDatabase(); this.isAsyncLoad = loadTsFileStatement.isAsyncLoad(); this.isVerifySchema = loadTsFileStatement.isVerifySchema(); + this.isAutoCreateSchemaAllowed = loadTsFileStatement.isAutoCreateSchema(); + this.isAutoCreateSchema = + IoTDBDescriptor.getInstance().getConfig().isAutoCreateSchemaEnabled() + && isAutoCreateSchemaAllowed; this.isAutoCreateDatabase = loadTsFileStatement.isAutoCreateDatabase(); this.isDeleteAfterLoad = loadTsFileStatement.isDeleteAfterLoad(); this.isConvertOnTypeMismatch = loadTsFileStatement.isConvertOnTypeMismatch(); @@ -165,6 +171,10 @@ public LoadTsFileAnalyzer( this.databaseForTableData = loadTsFileTableStatement.getDatabase(); this.isAsyncLoad = loadTsFileTableStatement.isAsyncLoad(); this.isVerifySchema = loadTsFileTableStatement.isVerifySchema(); + this.isAutoCreateSchemaAllowed = loadTsFileTableStatement.isAutoCreateSchema(); + this.isAutoCreateSchema = + IoTDBDescriptor.getInstance().getConfig().isAutoCreateSchemaEnabled() + && isAutoCreateSchemaAllowed; this.isAutoCreateDatabase = loadTsFileTableStatement.isAutoCreateDatabase(); this.isDeleteAfterLoad = loadTsFileTableStatement.isDeleteAfterLoad(); this.isConvertOnTypeMismatch = loadTsFileTableStatement.isConvertOnTypeMismatch(); @@ -188,6 +198,14 @@ protected boolean isAutoCreateDatabase() { return isAutoCreateDatabase; } + protected boolean isAutoCreateSchema() { + return isAutoCreateSchema; + } + + protected boolean isAutoCreateSchemaAllowed() { + return isAutoCreateSchemaAllowed; + } + protected boolean isConvertOnTypeMismatch() { return isConvertOnTypeMismatch; } @@ -298,6 +316,7 @@ private boolean doAsyncLoad(final IAnalysis analysis) { databaseLevel, isConvertOnTypeMismatch, isVerifySchema, + isAutoCreateSchemaAllowed, tabletConversionThresholdBytes, isGeneratedByPipe, Objects.nonNull(context) ? context.getUsername() : null); @@ -536,8 +555,7 @@ private void doAnalyzeSingleTreeFile( getOrCreateTreeSchemaVerifier().setCurrentModificationsAndTimeIndex(tsFileResource, reader); - final boolean isAutoCreateSchemaOrVerifySchemaEnabled = - IoTDBDescriptor.getInstance().getConfig().isAutoCreateSchemaEnabled() || isVerifySchema(); + final boolean isAutoCreateSchemaOrVerifySchemaEnabled = isAutoCreateSchema || isVerifySchema(); while (timeseriesMetadataIterator.hasNext()) { final Map> device2TimeseriesMetadata = timeseriesMetadataIterator.next(); @@ -846,9 +864,7 @@ private boolean setTemporaryUnavailableStatusIfNecessary( } boolean isTemporaryUnavailableDueToPipeSchemaNotReady(final Throwable throwable) { - if (!isGeneratedByPipe - || !isVerifySchema - || IoTDBDescriptor.getInstance().getConfig().isAutoCreateSchemaEnabled()) { + if (!isGeneratedByPipe || !isVerifySchema || isAutoCreateSchema) { return false; } 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 6d1704c0cfd0a..5bb12270e0088 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 @@ -497,6 +497,7 @@ private ISchemaTree autoCreateSchema() throws IllegalPathException { encodingsList, compressionTypesList, isAlignedList, + loadTsFileAnalyzer.isAutoCreateSchemaAllowed(), loadTsFileAnalyzer.context); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/schema/ISchemaFetcher.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/schema/ISchemaFetcher.java index f56c3e2c92f52..25dd523f3f0de 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/schema/ISchemaFetcher.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/schema/ISchemaFetcher.java @@ -24,6 +24,8 @@ import org.apache.iotdb.commons.schema.template.Template; import org.apache.iotdb.db.queryengine.common.MPPQueryContext; import org.apache.iotdb.db.queryengine.common.schematree.ISchemaTree; +import org.apache.iotdb.db.queryengine.plan.analyze.lock.DataNodeSchemaLockManager; +import org.apache.iotdb.db.queryengine.plan.analyze.lock.SchemaLockType; import org.apache.tsfile.enums.TSDataType; import org.apache.tsfile.file.metadata.enums.CompressionType; @@ -121,6 +123,21 @@ ISchemaTree fetchSchemaListWithAutoCreate( List aligned, MPPQueryContext context); + default ISchemaTree fetchSchemaList( + final List devicePaths, + final List measurementsList, + final MPPQueryContext context) { + DataNodeSchemaLockManager.getInstance() + .takeReadLock(context, SchemaLockType.VALIDATE_VS_DELETION_TREE); + final PathPatternTree patternTree = new PathPatternTree(); + for (int i = 0; i < devicePaths.size(); i++) { + for (final String measurement : measurementsList.get(i)) { + patternTree.appendFullPath(devicePaths.get(i), measurement); + } + } + return fetchSchema(patternTree, true, context, true); + } + Pair checkTemplateSetInfo(PartialPath devicePath); Pair checkTemplateSetAndPreSetInfo( 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 ff6efc8b3c1fb..6123afb1d9e8f 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 @@ -94,7 +94,37 @@ public static ISchemaTree validate( List compressionTypes, List isAlignedList, MPPQueryContext context) { - return schemaFetcher.fetchSchemaListWithAutoCreate( - devicePaths, measurements, dataTypes, encodings, compressionTypes, isAlignedList, context); + return validate( + schemaFetcher, + devicePaths, + measurements, + dataTypes, + encodings, + compressionTypes, + isAlignedList, + true, + context); + } + + public static ISchemaTree validate( + final ISchemaFetcher schemaFetcher, + final List devicePaths, + final List measurements, + final List dataTypes, + final List encodings, + final List compressionTypes, + final List isAlignedList, + final boolean autoCreateSchema, + final MPPQueryContext context) { + return autoCreateSchema + ? schemaFetcher.fetchSchemaListWithAutoCreate( + devicePaths, + measurements, + dataTypes, + encodings, + compressionTypes, + isAlignedList, + context) + : schemaFetcher.fetchSchemaList(devicePaths, measurements, context); } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/LoadTsFile.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/LoadTsFile.java index 28ae4d7f08163..17f958249291e 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/LoadTsFile.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/LoadTsFile.java @@ -60,6 +60,7 @@ public class LoadTsFile extends Statement { private long tabletConversionThresholdBytes; private boolean autoCreateDatabase; private boolean verify; + private boolean autoCreateSchema; private boolean isAsyncLoad = false; private boolean isGeneratedByPipe = false; @@ -97,6 +98,7 @@ private LoadTsFile( IoTDBDescriptor.getInstance().getConfig().getLoadTabletConversionThresholdBytes(); this.autoCreateDatabase = IoTDBDescriptor.getInstance().getConfig().isAutoCreateSchemaEnabled(); this.verify = true; + this.autoCreateSchema = true; this.loadAttributes = loadAttributes == null ? Collections.emptyMap() : loadAttributes; initAttributes(); @@ -155,6 +157,10 @@ public boolean isVerifySchema() { return verify; } + public boolean isAutoCreateSchema() { + return autoCreateSchema; + } + public int getDatabaseLevel() { return databaseLevel; } @@ -225,6 +231,8 @@ private void initAttributes() { this.tabletConversionThresholdBytes = LoadTsFileConfigurator.parseOrGetDefaultTabletConversionThresholdBytes(loadAttributes); this.verify = LoadTsFileConfigurator.parseOrGetDefaultVerify(loadAttributes); + this.autoCreateSchema = + LoadTsFileConfigurator.parseOrGetDefaultAutoCreateSchema(loadAttributes); this.isAsyncLoad = LoadTsFileConfigurator.parseOrGetDefaultAsyncLoad(loadAttributes); } @@ -304,6 +312,7 @@ public List getSubStatements() { subStatement.databaseLevel = this.databaseLevel; subStatement.database = this.database; subStatement.verify = this.verify; + subStatement.autoCreateSchema = this.autoCreateSchema; subStatement.deleteAfterLoad = this.deleteAfterLoad; subStatement.convertOnTypeMismatch = this.convertOnTypeMismatch; subStatement.tabletConversionThresholdBytes = this.tabletConversionThresholdBytes; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/crud/LoadTsFileStatement.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/crud/LoadTsFileStatement.java index 0d632ef87bf7e..ca823fcb5579e 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/crud/LoadTsFileStatement.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/crud/LoadTsFileStatement.java @@ -49,6 +49,7 @@ import static org.apache.iotdb.commons.conf.IoTDBConstant.PATH_ROOT; 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.CONVERT_ON_TYPE_MISMATCH_KEY; import static org.apache.iotdb.db.storageengine.load.config.LoadTsFileConfigurator.DATABASE_LEVEL_KEY; import static org.apache.iotdb.db.storageengine.load.config.LoadTsFileConfigurator.DATABASE_NAME_KEY; @@ -64,6 +65,7 @@ public class LoadTsFileStatement extends Statement { private int databaseLevel; // For loading to tree-model only private String database; // For loading to table-model only private boolean verifySchema = true; + private boolean autoCreateSchema = true; private boolean deleteAfterLoad = false; private boolean convertOnTypeMismatch = true; private long tabletConversionThresholdBytes = -1; @@ -90,6 +92,7 @@ private LoadTsFileStatement(String filePath, boolean validateSourcePath) this.file = new File(filePath).getAbsoluteFile(); this.databaseLevel = IoTDBDescriptor.getInstance().getConfig().getDefaultDatabaseLevel(); this.verifySchema = true; + this.autoCreateSchema = true; this.deleteAfterLoad = false; this.convertOnTypeMismatch = true; this.tabletConversionThresholdBytes = @@ -134,6 +137,7 @@ protected LoadTsFileStatement() { this.file = null; this.databaseLevel = IoTDBDescriptor.getInstance().getConfig().getDefaultDatabaseLevel(); this.verifySchema = true; + this.autoCreateSchema = true; this.deleteAfterLoad = false; this.convertOnTypeMismatch = true; this.tabletConversionThresholdBytes = @@ -245,6 +249,14 @@ public boolean isVerifySchema() { return verifySchema; } + public void setAutoCreateSchema(final boolean autoCreateSchema) { + this.autoCreateSchema = autoCreateSchema; + } + + public boolean isAutoCreateSchema() { + return autoCreateSchema; + } + public LoadTsFileStatement setDeleteAfterLoad(boolean deleteAfterLoad) { this.deleteAfterLoad = deleteAfterLoad; return this; @@ -340,6 +352,8 @@ private void initAttributes(final Map loadAttributes) { this.tabletConversionThresholdBytes = LoadTsFileConfigurator.parseOrGetDefaultTabletConversionThresholdBytes(loadAttributes); this.verifySchema = LoadTsFileConfigurator.parseOrGetDefaultVerify(loadAttributes); + this.autoCreateSchema = + LoadTsFileConfigurator.parseOrGetDefaultAutoCreateSchema(loadAttributes); this.isAsyncLoad = LoadTsFileConfigurator.parseOrGetDefaultAsyncLoad(loadAttributes); if (LoadTsFileConfigurator.parseOrGetDefaultPipeGenerated(loadAttributes)) { markIsGeneratedByPipe(); @@ -436,6 +450,7 @@ public List getSubStatements() { statement.databaseLevel = this.databaseLevel; statement.database = this.database; statement.verifySchema = this.verifySchema; + statement.autoCreateSchema = this.autoCreateSchema; statement.deleteAfterLoad = this.deleteAfterLoad; statement.convertOnTypeMismatch = this.convertOnTypeMismatch; statement.tabletConversionThresholdBytes = this.tabletConversionThresholdBytes; @@ -478,6 +493,7 @@ public List getPaths() { loadAttributes.put( TABLET_CONVERSION_THRESHOLD_KEY, String.valueOf(tabletConversionThresholdBytes)); loadAttributes.put(ASYNC_LOAD_KEY, String.valueOf(isAsyncLoad)); + loadAttributes.put(AUTO_CREATE_SCHEMA_KEY, String.valueOf(autoCreateSchema)); if (isGeneratedByPipe) { loadAttributes.put(PIPE_GENERATED_KEY, String.valueOf(true)); } @@ -501,6 +517,8 @@ public String toString() { + databaseLevel + ", verify-schema=" + verifySchema + + ", auto-create-schema=" + + autoCreateSchema + ", convert-on-type-mismatch=" + convertOnTypeMismatch + ", tablet-conversion-threshold=" diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadPathHelper.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadPathHelper.java index 991d368d39530..7792cae1a77dc 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadPathHelper.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadPathHelper.java @@ -59,6 +59,7 @@ public final class ActiveLoadPathHelper { LoadTsFileConfigurator.CONVERT_ON_TYPE_MISMATCH_KEY, LoadTsFileConfigurator.TABLET_CONVERSION_THRESHOLD_KEY, LoadTsFileConfigurator.VERIFY_KEY, + LoadTsFileConfigurator.AUTO_CREATE_SCHEMA_KEY, LoadTsFileConfigurator.DATABASE_KEY, LoadTsFileConfigurator.PIPE_GENERATED_KEY)); @@ -71,6 +72,7 @@ public static Map buildAttributes( final Integer databaseLevel, final Boolean convertOnTypeMismatch, final Boolean verify, + final Boolean autoCreateSchema, final Long tabletConversionThresholdBytes, final Boolean pipeGenerated, final String userName) { @@ -103,6 +105,11 @@ public static Map buildAttributes( attributes.put(LoadTsFileConfigurator.VERIFY_KEY, Boolean.toString(verify)); } + if (Objects.nonNull(autoCreateSchema)) { + attributes.put( + LoadTsFileConfigurator.AUTO_CREATE_SCHEMA_KEY, Boolean.toString(autoCreateSchema)); + } + if (Objects.nonNull(pipeGenerated) && pipeGenerated) { attributes.put(LoadTsFileConfigurator.PIPE_GENERATED_KEY, Boolean.TRUE.toString()); } @@ -208,6 +215,9 @@ public static void applyAttributesToStatement( statement.setVerifySchema(defaultVerify); } + Optional.ofNullable(attributes.get(LoadTsFileConfigurator.AUTO_CREATE_SCHEMA_KEY)) + .ifPresent(value -> statement.setAutoCreateSchema(Boolean.parseBoolean(value))); + if (attributes.containsKey(LoadTsFileConfigurator.PIPE_GENERATED_KEY) && Boolean.parseBoolean(attributes.get(LoadTsFileConfigurator.PIPE_GENERATED_KEY))) { statement.markIsGeneratedByPipe(); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/config/LoadTsFileConfigurator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/config/LoadTsFileConfigurator.java index 3076afe1f2551..8d4755aaf7025 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/config/LoadTsFileConfigurator.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/config/LoadTsFileConfigurator.java @@ -55,6 +55,9 @@ public static void validateParameters(final String key, final String value) { case VERIFY_KEY: validateVerifyParam(value); break; + case AUTO_CREATE_SCHEMA_KEY: + validateAutoCreateSchemaParam(value); + break; case PIPE_GENERATED_KEY: validatePipeGeneratedParam(value); break; @@ -184,6 +187,26 @@ public static boolean parseOrGetDefaultVerify(final Map loadAttr loadAttributes.getOrDefault(VERIFY_KEY, String.valueOf(VERIFY_DEFAULT_VALUE))); } + public static final String AUTO_CREATE_SCHEMA_KEY = "auto-create-schema"; + private static final boolean AUTO_CREATE_SCHEMA_DEFAULT_VALUE = true; + + public static void validateAutoCreateSchemaParam(final String autoCreateSchema) { + if (!"true".equalsIgnoreCase(autoCreateSchema) && !"false".equalsIgnoreCase(autoCreateSchema)) { + throw new SemanticException( + String.format( + StorageEngineMessages.PARAMETER_VALUE_NOT_SUPPORTED_BOOLEAN, + AUTO_CREATE_SCHEMA_KEY, + autoCreateSchema)); + } + } + + public static boolean parseOrGetDefaultAutoCreateSchema( + final Map loadAttributes) { + return Boolean.parseBoolean( + loadAttributes.getOrDefault( + AUTO_CREATE_SCHEMA_KEY, String.valueOf(AUTO_CREATE_SCHEMA_DEFAULT_VALUE))); + } + public static final String PIPE_GENERATED_KEY = "pipe-generated"; public static void validatePipeGeneratedParam(final String pipeGenerated) { diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/builder/PipeDataNodeTaskBuilderTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/builder/PipeDataNodeTaskBuilderTest.java index e00f9500bbef7..f3ce153970dd0 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/builder/PipeDataNodeTaskBuilderTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/builder/PipeDataNodeTaskBuilderTest.java @@ -31,6 +31,7 @@ import org.junit.Assert; import org.junit.Test; +import java.util.Arrays; import java.util.HashMap; import java.util.Map; @@ -108,4 +109,82 @@ public void testPreprocessParametersInjectsEventUserForExternalWriteBackSink() { Boolean.TRUE.toString(), sinkParameters.getStringByKeys(PipeSinkConstant.CONNECTOR_USE_EVENT_USER_NAME_KEY)); } + + @Test + public void testPreprocessParametersWaitsForCompleteIoTDBSchemaHistory() { + final Map sourceAttributes = new HashMap<>(); + sourceAttributes.put(PipeSourceConstant.SOURCE_INCLUSION_KEY, "all"); + final PipeParameters sinkParameters = new PipeParameters(new HashMap<>()); + + PipeDataNodeTaskBuilder.preprocessParameters( + new PipeParameters(sourceAttributes), sinkParameters); + + Assert.assertEquals( + Boolean.TRUE.toString(), + sinkParameters.getStringByKeys(SystemConstant.SINK_WAIT_FOR_SCHEMA_BEFORE_LOAD_KEY)); + } + + @Test + public void testPreprocessParametersDoesNotWaitForIncompleteSchemaHistory() { + for (final String excludedOption : + Arrays.asList("schema.timeseries.template.alter", "schema.timeseries.template.activate")) { + final Map sourceAttributes = new HashMap<>(); + sourceAttributes.put(PipeSourceConstant.SOURCE_INCLUSION_KEY, "all"); + sourceAttributes.put(PipeSourceConstant.SOURCE_EXCLUSION_KEY, excludedOption); + final Map sinkAttributes = new HashMap<>(); + sinkAttributes.put( + SystemConstant.SINK_WAIT_FOR_SCHEMA_BEFORE_LOAD_KEY, Boolean.TRUE.toString()); + final PipeParameters sinkParameters = new PipeParameters(sinkAttributes); + + PipeDataNodeTaskBuilder.preprocessParameters( + new PipeParameters(sourceAttributes), sinkParameters); + + Assert.assertEquals( + Boolean.FALSE.toString(), + sinkParameters.getStringByKeys(SystemConstant.SINK_WAIT_FOR_SCHEMA_BEFORE_LOAD_KEY)); + } + } + + @Test + public void testPreprocessParametersDoesNotWaitForExternalSourceOrGeneralWrite() { + final Map externalSourceAttributes = new HashMap<>(); + externalSourceAttributes.put(PipeSourceConstant.SOURCE_KEY, "external-source"); + externalSourceAttributes.put(PipeSourceConstant.SOURCE_INCLUSION_KEY, "all"); + final PipeParameters externalSourceSinkParameters = new PipeParameters(new HashMap<>()); + + PipeDataNodeTaskBuilder.preprocessParameters( + new PipeParameters(externalSourceAttributes), externalSourceSinkParameters); + + Assert.assertEquals( + Boolean.FALSE.toString(), + externalSourceSinkParameters.getStringByKeys( + SystemConstant.SINK_WAIT_FOR_SCHEMA_BEFORE_LOAD_KEY)); + + final Map sourceAttributes = new HashMap<>(); + sourceAttributes.put(PipeSourceConstant.SOURCE_INCLUSION_KEY, "all"); + final Map sinkAttributes = new HashMap<>(); + sinkAttributes.put( + PipeSinkConstant.SINK_MARK_AS_GENERAL_WRITE_REQUEST_KEY, Boolean.TRUE.toString()); + final PipeParameters generalWriteSinkParameters = new PipeParameters(sinkAttributes); + + PipeDataNodeTaskBuilder.preprocessParameters( + new PipeParameters(sourceAttributes), generalWriteSinkParameters); + + Assert.assertEquals( + Boolean.FALSE.toString(), + generalWriteSinkParameters.getStringByKeys( + SystemConstant.SINK_WAIT_FOR_SCHEMA_BEFORE_LOAD_KEY)); + + final Map nonPipeSinkAttributes = new HashMap<>(); + nonPipeSinkAttributes.put( + PipeSinkConstant.SINK_MARK_AS_PIPE_REQUEST_KEY, Boolean.FALSE.toString()); + final PipeParameters nonPipeSinkParameters = new PipeParameters(nonPipeSinkAttributes); + + PipeDataNodeTaskBuilder.preprocessParameters( + new PipeParameters(sourceAttributes), nonPipeSinkParameters); + + Assert.assertEquals( + Boolean.FALSE.toString(), + nonPipeSinkParameters.getStringByKeys(SystemConstant.SINK_WAIT_FOR_SCHEMA_BEFORE_LOAD_KEY)); + } } diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiverTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiverTest.java index 8f2e86c62d057..79576ed2680fd 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiverTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiverTest.java @@ -71,6 +71,36 @@ public void testLoadTsFileAsyncAttributesUseTreeDatabaseLevelFromDatabaseName() Assert.assertEquals("root.test.sg_0", statement.getDatabase()); Assert.assertEquals(2, statement.getDatabaseLevel()); Assert.assertTrue(statement.isVerifySchema()); + Assert.assertTrue(statement.isAutoCreateSchema()); + } finally { + Files.deleteIfExists(tsFile); + } + } + + @Test + public void testLoadTsFileWaitsForSchemaInSyncAndAsyncModes() throws Exception { + final Path tsFile = Files.createTempFile("pipe-load-wait-for-schema", ".tsfile"); + try { + final LoadTsFileStatement syncStatement = + IoTDBDataNodeReceiver.buildLoadTsFileStatementForSync( + "root.test.sg_0", tsFile.toString(), false, false, true); + Assert.assertTrue(syncStatement.isVerifySchema()); + Assert.assertFalse(syncStatement.isAutoCreateSchema()); + + final Map asyncAttributes = + IoTDBDataNodeReceiver.buildLoadTsFileAttributesForAsync( + "root.test.sg_0", false, false, true, true); + Assert.assertEquals( + Boolean.TRUE.toString(), asyncAttributes.get(LoadTsFileConfigurator.VERIFY_KEY)); + Assert.assertEquals( + Boolean.FALSE.toString(), + asyncAttributes.get(LoadTsFileConfigurator.AUTO_CREATE_SCHEMA_KEY)); + + final LoadTsFileStatement asyncStatement = + LoadTsFileStatement.createUnchecked(tsFile.toString()); + ActiveLoadPathHelper.applyAttributesToStatement(asyncAttributes, asyncStatement, false); + Assert.assertTrue(asyncStatement.isVerifySchema()); + Assert.assertFalse(asyncStatement.isAutoCreateSchema()); } finally { Files.deleteIfExists(tsFile); } diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/PipeDataNodeThriftRequestTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/PipeDataNodeThriftRequestTest.java index 07b7661e2cb6e..92d3e0e687722 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/PipeDataNodeThriftRequestTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/PipeDataNodeThriftRequestTest.java @@ -1177,6 +1177,18 @@ public void testPipeTransferTsFileSealWithModReq() throws IOException { Assert.assertEquals(Arrays.asList(modFileName, tsFileName), deserializeReq.getFileNames()); Assert.assertEquals(Arrays.asList(10L, 100L), deserializeReq.getFileLengths()); Assert.assertEquals("root.db", deserializeReq.getDatabaseNameByTsFileName()); + Assert.assertFalse(deserializeReq.shouldWaitForSchemaBeforeLoad()); + } + + @Test + public void testPipeTransferTsFileSealWithModReqWaitsForSchema() throws IOException { + final PipeTransferTsFileSealWithModReq req = + PipeTransferTsFileSealWithModReq.toTPipeTransferReq( + "1.tsfile.mod", 10, "1.tsfile", 100, "root.db", true); + final PipeTransferTsFileSealWithModReq deserializeReq = + PipeTransferTsFileSealWithModReq.fromTPipeTransferReq(req); + + Assert.assertTrue(deserializeReq.shouldWaitForSchemaBeforeLoad()); } @Test @@ -1201,6 +1213,7 @@ public void testPipeTransferTsFileSealWithModReqFromLegacyV13BodyWithoutDatabase Assert.assertEquals(Arrays.asList(10L, 100L), deserializeReq.getFileLengths()); Assert.assertTrue(deserializeReq.getParameters().isEmpty()); Assert.assertNull(deserializeReq.getDatabaseNameByTsFileName()); + Assert.assertFalse(deserializeReq.shouldWaitForSchemaBeforeLoad()); } @Test 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 68f4bde29e6a6..cda3515db9e68 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 @@ -231,6 +231,67 @@ public void testPipeGeneratedLoadMissingSchemaShouldBeTemporaryWhenAutoCreateDis } } + @Test + public void testPipeGeneratedLoadMissingSchemaShouldBeTemporaryWhenPerLoadAutoCreateDisabled() + throws Exception { + final boolean originalAutoCreateSchemaEnabled = + IoTDBDescriptor.getInstance().getConfig().isAutoCreateSchemaEnabled(); + IoTDBDescriptor.getInstance().getConfig().setAutoCreateSchemaEnabled(true); + final File tsFile = File.createTempFile("missing-schema-per-load", ".tsfile"); + + try { + final LoadTsFileStatement waitingStatement = + LoadTsFileStatement.createUnchecked(tsFile.getAbsolutePath()); + waitingStatement.setAutoCreateSchema(false); + try (final LoadTsFileAnalyzer waitingAnalyzer = + new LoadTsFileAnalyzer( + waitingStatement, true, new MPPQueryContext(new QueryId("load_pipe_waiting_test")))) { + Assert.assertFalse(waitingAnalyzer.isAutoCreateSchemaAllowed()); + Assert.assertTrue( + waitingAnalyzer.isTemporaryUnavailableDueToPipeSchemaNotReady( + new LoadAnalyzeMissingSchemaException("missing schema"))); + } + + try (final LoadTsFileAnalyzer defaultAnalyzer = + new LoadTsFileAnalyzer( + LoadTsFileStatement.createUnchecked(tsFile.getAbsolutePath()), + true, + new MPPQueryContext(new QueryId("load_pipe_default_test")))) { + Assert.assertTrue(defaultAnalyzer.isAutoCreateSchemaAllowed()); + Assert.assertFalse( + defaultAnalyzer.isTemporaryUnavailableDueToPipeSchemaNotReady( + new LoadAnalyzeMissingSchemaException("missing schema"))); + } + } finally { + IoTDBDescriptor.getInstance() + .getConfig() + .setAutoCreateSchemaEnabled(originalAutoCreateSchemaEnabled); + Assert.assertTrue(tsFile.delete()); + } + } + + @Test + public void testGlobalAutoCreateDisabledKeepsPerLoadAutoCreatePermission() throws Exception { + final boolean originalAutoCreateSchemaEnabled = + IoTDBDescriptor.getInstance().getConfig().isAutoCreateSchemaEnabled(); + IoTDBDescriptor.getInstance().getConfig().setAutoCreateSchemaEnabled(false); + final File tsFile = File.createTempFile("global-auto-create-disabled", ".tsfile"); + + try (final LoadTsFileAnalyzer analyzer = + new LoadTsFileAnalyzer( + LoadTsFileStatement.createUnchecked(tsFile.getAbsolutePath()), + true, + new MPPQueryContext(new QueryId("load_global_auto_create_disabled_test")))) { + Assert.assertFalse(analyzer.isAutoCreateSchema()); + Assert.assertTrue(analyzer.isAutoCreateSchemaAllowed()); + } finally { + IoTDBDescriptor.getInstance() + .getConfig() + .setAutoCreateSchemaEnabled(originalAutoCreateSchemaEnabled); + Assert.assertTrue(tsFile.delete()); + } + } + private void writeTableTsFileWithMixedDevices(final File tsFile) throws Exception { if (tsFile.exists()) { Assert.assertTrue(tsFile.delete()); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadDirScannerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadDirScannerTest.java index 60532f749d8e0..1669555334027 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadDirScannerTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadDirScannerTest.java @@ -111,7 +111,7 @@ public void testAttributeAndTransferDirectoriesDoNotImplyTableModel() throws Exc // Async tree loads add attribute and per-handoff transfer directories below pending. These are // internal directories, not table database names inferred from a user-created subdirectory. final Map attributes = - ActiveLoadPathHelper.buildAttributes(null, 2, false, false, null, false, "test-user"); + ActiveLoadPathHelper.buildAttributes(null, 2, false, false, null, null, false, "test-user"); final File attributeDir = ActiveLoadPathHelper.resolveTargetDir(pendingDir, attributes); final File transferDir = new File(attributeDir, "transfer-id"); Assert.assertTrue(transferDir.mkdirs()); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadPathHelperTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadPathHelperTest.java index 77936a653c7b5..c339aef5977aa 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadPathHelperTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadPathHelperTest.java @@ -39,7 +39,8 @@ public void testUserAttributeShouldBeMaskedInPathAndDecodedWhenParsing() throws final File targetDir = ActiveLoadPathHelper.resolveTargetDir( pendingDir, - ActiveLoadPathHelper.buildAttributes(null, null, null, null, null, null, userName)); + ActiveLoadPathHelper.buildAttributes( + null, null, null, null, null, null, null, userName)); final File tsFile = new File(targetDir, "1-0-0-0.tsfile"); Assert.assertTrue(targetDir.getAbsolutePath().contains("user-v1-")); @@ -126,6 +127,32 @@ public void testKnownPrefixWithInvalidFutureLikeValueShouldBeIgnoredForDowngrade } } + @Test + public void testAutoCreateSchemaAttributeShouldSurviveActiveLoadPath() throws Exception { + final File pendingDir = Files.createTempDirectory("active-load-schema").toFile(); + try { + final Map attributes = + ActiveLoadPathHelper.buildAttributes(null, null, null, true, false, null, true, "root"); + final File targetDir = ActiveLoadPathHelper.resolveTargetDir(pendingDir, attributes); + final File tsFile = new File(targetDir, "1-0-0-0.tsfile"); + createFile(tsFile); + + final Map parsedAttributes = + ActiveLoadPathHelper.parseAttributes(tsFile, pendingDir); + Assert.assertEquals( + Boolean.FALSE.toString(), + parsedAttributes.get(LoadTsFileConfigurator.AUTO_CREATE_SCHEMA_KEY)); + + final LoadTsFileStatement statement = + LoadTsFileStatement.createUnchecked(tsFile.getAbsolutePath()); + ActiveLoadPathHelper.applyAttributesToStatement(parsedAttributes, statement, false); + Assert.assertTrue(statement.isVerifySchema()); + Assert.assertFalse(statement.isAutoCreateSchema()); + } finally { + deleteRecursively(pendingDir); + } + } + private static void deleteRecursively(final File file) { if (file == null || !file.exists()) { return; diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/SystemConstant.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/SystemConstant.java index 152b8c008f3e7..f1ca66be0988e 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/SystemConstant.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/SystemConstant.java @@ -48,6 +48,8 @@ public class SystemConstant { "__system.source-authentication-injected"; public static final String SINK_AUTHENTICATION_INJECTED_KEY = "__system.sink-authentication-injected"; + public static final String SINK_WAIT_FOR_SCHEMA_BEFORE_LOAD_KEY = + "__system.sink-wait-for-schema-before-load"; /////////////////////////////////// Utility /////////////////////////////////// @@ -59,6 +61,7 @@ public class SystemConstant { SYSTEM_KEYS.add(PIPE_VISIBILITY_KEY); SYSTEM_KEYS.add(SOURCE_AUTHENTICATION_INJECTED_KEY); SYSTEM_KEYS.add(SINK_AUTHENTICATION_INJECTED_KEY); + SYSTEM_KEYS.add(SINK_WAIT_FOR_SCHEMA_BEFORE_LOAD_KEY); } public static PipeParameters addSystemKeysIfNecessary(final PipeParameters givenPipeParameters) { diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/options/PipeInclusionOptions.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/options/PipeInclusionOptions.java index 9a4f2d09ce755..3bde97c1a12ae 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/options/PipeInclusionOptions.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/options/PipeInclusionOptions.java @@ -257,6 +257,21 @@ public static Set parseOptions(final String optionsString) return options; } + public static boolean areOptionsEnabled(final PipeParameters parameters, final String... options) + throws IllegalPathException { + final Set inclusionOptions = parseOptions(getInclusionString(parameters)); + final Set exclusionOptions = parseOptions(getExclusionString(parameters)); + + for (final String option : options) { + final PartialPath optionPath = new PartialPath(option); + if (inclusionOptions.stream().noneMatch(optionPath::matchPrefixPath) + || exclusionOptions.stream().anyMatch(optionPath::matchPrefixPath)) { + return false; + } + } + return true; + } + private PipeInclusionOptions() { // Utility class } diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/protocol/IoTDBSink.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/protocol/IoTDBSink.java index 9362b62f4143f..eb2fdfd73d0c6 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/protocol/IoTDBSink.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/protocol/IoTDBSink.java @@ -23,6 +23,7 @@ import org.apache.iotdb.commons.audit.UserEntity; import org.apache.iotdb.commons.i18n.PipeMessages; import org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant; +import org.apache.iotdb.commons.pipe.config.constant.SystemConstant; import org.apache.iotdb.commons.pipe.config.plugin.env.PipeTaskSinkRuntimeEnvironment; import org.apache.iotdb.commons.pipe.receiver.PipeReceiverStatusHandler; import org.apache.iotdb.commons.pipe.sink.compressor.PipeCompressor; @@ -168,6 +169,7 @@ public abstract class IoTDBSink implements PipeConnector, PipeConnectorWithEvent protected String loadTsFileStrategy; protected boolean loadTsFileValidation; + protected boolean shouldWaitForSchemaBeforeLoad; protected boolean shouldMarkAsPipeRequest; protected boolean skipIfNoPrivileges; @@ -302,6 +304,8 @@ public void validate(final PipeParameterValidator validator) throws Exception { parameters.getBooleanOrDefault( Arrays.asList(CONNECTOR_LOAD_TSFILE_VALIDATION_KEY, SINK_LOAD_TSFILE_VALIDATION_KEY), CONNECTOR_LOAD_TSFILE_VALIDATION_DEFAULT_VALUE); + shouldWaitForSchemaBeforeLoad = + parameters.getBooleanOrDefault(SystemConstant.SINK_WAIT_FOR_SCHEMA_BEFORE_LOAD_KEY, false); final int zstdCompressionLevel = parameters.getIntOrDefault( @@ -656,6 +660,10 @@ public PipeReceiverStatusHandler statusHandler() { return receiverStatusHandler; } + public boolean shouldWaitForSchemaBeforeLoad() { + return shouldWaitForSchemaBeforeLoad; + } + public void setTabletBatchSizeHistogram(Histogram tabletBatchSizeHistogram) { // do nothing by default } From 4ee2fbef3b418bc5d323673b2ff559cac8e9bf76 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Thu, 30 Jul 2026 14:25:39 +0800 Subject: [PATCH 2/2] Clarify load schema creation and lock lifecycle --- .../plan/analyze/load/LoadTsFileAnalyzer.java | 31 ++++++++++--------- .../TreeSchemaAutoCreatorAndVerifier.java | 2 +- .../plan/analyze/schema/ISchemaFetcher.java | 2 ++ .../analyze/load/LoadTsFileAnalyzerTest.java | 8 ++--- 4 files changed, 23 insertions(+), 20 deletions(-) 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 aa274a100c049..34827de0c63da 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 @@ -114,8 +114,8 @@ public class LoadTsFileAnalyzer implements AutoCloseable { private String databaseForTableData; private final boolean isAsyncLoad; private final boolean isVerifySchema; - private final boolean isAutoCreateSchemaAllowed; - private final boolean isAutoCreateSchema; + private final boolean isAutoCreateSchemaRequested; + private final boolean isAutoCreateSchemaEnabled; private final boolean isAutoCreateDatabase; private final boolean isDeleteAfterLoad; private final boolean isConvertOnTypeMismatch; @@ -143,10 +143,10 @@ public LoadTsFileAnalyzer( this.databaseForTableData = loadTsFileStatement.getDatabase(); this.isAsyncLoad = loadTsFileStatement.isAsyncLoad(); this.isVerifySchema = loadTsFileStatement.isVerifySchema(); - this.isAutoCreateSchemaAllowed = loadTsFileStatement.isAutoCreateSchema(); - this.isAutoCreateSchema = + this.isAutoCreateSchemaRequested = loadTsFileStatement.isAutoCreateSchema(); + this.isAutoCreateSchemaEnabled = IoTDBDescriptor.getInstance().getConfig().isAutoCreateSchemaEnabled() - && isAutoCreateSchemaAllowed; + && isAutoCreateSchemaRequested; this.isAutoCreateDatabase = loadTsFileStatement.isAutoCreateDatabase(); this.isDeleteAfterLoad = loadTsFileStatement.isDeleteAfterLoad(); this.isConvertOnTypeMismatch = loadTsFileStatement.isConvertOnTypeMismatch(); @@ -171,10 +171,10 @@ public LoadTsFileAnalyzer( this.databaseForTableData = loadTsFileTableStatement.getDatabase(); this.isAsyncLoad = loadTsFileTableStatement.isAsyncLoad(); this.isVerifySchema = loadTsFileTableStatement.isVerifySchema(); - this.isAutoCreateSchemaAllowed = loadTsFileTableStatement.isAutoCreateSchema(); - this.isAutoCreateSchema = + this.isAutoCreateSchemaRequested = loadTsFileTableStatement.isAutoCreateSchema(); + this.isAutoCreateSchemaEnabled = IoTDBDescriptor.getInstance().getConfig().isAutoCreateSchemaEnabled() - && isAutoCreateSchemaAllowed; + && isAutoCreateSchemaRequested; this.isAutoCreateDatabase = loadTsFileTableStatement.isAutoCreateDatabase(); this.isDeleteAfterLoad = loadTsFileTableStatement.isDeleteAfterLoad(); this.isConvertOnTypeMismatch = loadTsFileTableStatement.isConvertOnTypeMismatch(); @@ -198,12 +198,12 @@ protected boolean isAutoCreateDatabase() { return isAutoCreateDatabase; } - protected boolean isAutoCreateSchema() { - return isAutoCreateSchema; + protected boolean isAutoCreateSchemaEnabled() { + return isAutoCreateSchemaEnabled; } - protected boolean isAutoCreateSchemaAllowed() { - return isAutoCreateSchemaAllowed; + protected boolean isAutoCreateSchemaRequested() { + return isAutoCreateSchemaRequested; } protected boolean isConvertOnTypeMismatch() { @@ -316,7 +316,7 @@ private boolean doAsyncLoad(final IAnalysis analysis) { databaseLevel, isConvertOnTypeMismatch, isVerifySchema, - isAutoCreateSchemaAllowed, + isAutoCreateSchemaRequested, tabletConversionThresholdBytes, isGeneratedByPipe, Objects.nonNull(context) ? context.getUsername() : null); @@ -555,7 +555,8 @@ private void doAnalyzeSingleTreeFile( getOrCreateTreeSchemaVerifier().setCurrentModificationsAndTimeIndex(tsFileResource, reader); - final boolean isAutoCreateSchemaOrVerifySchemaEnabled = isAutoCreateSchema || isVerifySchema(); + final boolean isAutoCreateSchemaOrVerifySchemaEnabled = + isAutoCreateSchemaEnabled || isVerifySchema(); while (timeseriesMetadataIterator.hasNext()) { final Map> device2TimeseriesMetadata = timeseriesMetadataIterator.next(); @@ -864,7 +865,7 @@ private boolean setTemporaryUnavailableStatusIfNecessary( } boolean isTemporaryUnavailableDueToPipeSchemaNotReady(final Throwable throwable) { - if (!isGeneratedByPipe || !isVerifySchema || isAutoCreateSchema) { + if (!isGeneratedByPipe || !isVerifySchema || isAutoCreateSchemaEnabled) { return false; } 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 5bb12270e0088..3b04660a415e6 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 @@ -497,7 +497,7 @@ private ISchemaTree autoCreateSchema() throws IllegalPathException { encodingsList, compressionTypesList, isAlignedList, - loadTsFileAnalyzer.isAutoCreateSchemaAllowed(), + loadTsFileAnalyzer.isAutoCreateSchemaRequested(), loadTsFileAnalyzer.context); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/schema/ISchemaFetcher.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/schema/ISchemaFetcher.java index 25dd523f3f0de..7cf73c942ad6f 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/schema/ISchemaFetcher.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/schema/ISchemaFetcher.java @@ -127,6 +127,8 @@ default ISchemaTree fetchSchemaList( final List devicePaths, final List measurementsList, final MPPQueryContext context) { + // The lock is recorded in the query context and intentionally remains held through execution. + // The query lifecycle releases all recorded schema read locks in Coordinator's finally block. DataNodeSchemaLockManager.getInstance() .takeReadLock(context, SchemaLockType.VALIDATE_VS_DELETION_TREE); final PathPatternTree patternTree = new PathPatternTree(); 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 cda3515db9e68..d3753a051de8c 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 @@ -246,7 +246,7 @@ public void testPipeGeneratedLoadMissingSchemaShouldBeTemporaryWhenPerLoadAutoCr try (final LoadTsFileAnalyzer waitingAnalyzer = new LoadTsFileAnalyzer( waitingStatement, true, new MPPQueryContext(new QueryId("load_pipe_waiting_test")))) { - Assert.assertFalse(waitingAnalyzer.isAutoCreateSchemaAllowed()); + Assert.assertFalse(waitingAnalyzer.isAutoCreateSchemaRequested()); Assert.assertTrue( waitingAnalyzer.isTemporaryUnavailableDueToPipeSchemaNotReady( new LoadAnalyzeMissingSchemaException("missing schema"))); @@ -257,7 +257,7 @@ waitingStatement, true, new MPPQueryContext(new QueryId("load_pipe_waiting_test" LoadTsFileStatement.createUnchecked(tsFile.getAbsolutePath()), true, new MPPQueryContext(new QueryId("load_pipe_default_test")))) { - Assert.assertTrue(defaultAnalyzer.isAutoCreateSchemaAllowed()); + Assert.assertTrue(defaultAnalyzer.isAutoCreateSchemaRequested()); Assert.assertFalse( defaultAnalyzer.isTemporaryUnavailableDueToPipeSchemaNotReady( new LoadAnalyzeMissingSchemaException("missing schema"))); @@ -282,8 +282,8 @@ public void testGlobalAutoCreateDisabledKeepsPerLoadAutoCreatePermission() throw LoadTsFileStatement.createUnchecked(tsFile.getAbsolutePath()), true, new MPPQueryContext(new QueryId("load_global_auto_create_disabled_test")))) { - Assert.assertFalse(analyzer.isAutoCreateSchema()); - Assert.assertTrue(analyzer.isAutoCreateSchemaAllowed()); + Assert.assertFalse(analyzer.isAutoCreateSchemaEnabled()); + Assert.assertTrue(analyzer.isAutoCreateSchemaRequested()); } finally { IoTDBDescriptor.getInstance() .getConfig()