This is an automated email from the ASF dual-hosted git repository. Caideyipi pushed a commit to branch fix/pipe-historical-schema-race in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 4cc75a8e74fc83910c74dc37e9ed2364dce31f9e Author: Caideyipi <[email protected]> AuthorDate: Mon Jul 27 15:09:35 2026 +0800 [Pipe] Fix historical TsFile schema recovery race --- .../auto/enhanced/IoTDBPipeAutoConflictIT.java | 5 ++ .../task/builder/PipeDataNodeTaskBuilder.java | 56 +++++++++++++-- .../protocol/thrift/IoTDBDataNodeReceiver.java | 76 ++++++++++++++++----- .../request/PipeTransferTsFileSealWithModReq.java | 72 +++++++++++++++++--- .../protocol/airgap/IoTDBDataRegionAirGapSink.java | 9 ++- .../async/handler/PipeTransferTsFileHandler.java | 8 ++- .../thrift/sync/IoTDBDataRegionSyncSink.java | 8 ++- .../plan/analyze/load/LoadTsFileAnalyzer.java | 26 +++++-- .../load/TreeSchemaAutoCreatorAndVerifier.java | 1 + .../plan/analyze/schema/ISchemaFetcher.java | 17 +++++ .../plan/analyze/schema/SchemaValidator.java | 34 +++++++++- .../plan/relational/sql/ast/LoadTsFile.java | 9 +++ .../plan/statement/crud/LoadTsFileStatement.java | 18 +++++ .../load/active/ActiveLoadPathHelper.java | 10 +++ .../load/config/LoadTsFileConfigurator.java | 23 +++++++ .../task/builder/PipeDataNodeTaskBuilderTest.java | 79 ++++++++++++++++++++++ .../protocol/thrift/IoTDBDataNodeReceiverTest.java | 30 ++++++++ .../pipe/sink/PipeDataNodeThriftRequestTest.java | 13 ++++ .../plan/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 6953a5a23f4..8562194be99 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 class IoTDBPipeAutoConflictIT extends AbstractPipeDualTreeModelAutoIT { 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 98157a2a49a..ae4ef473998 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.PipeSinkConstant.SIN 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 @@ public class PipeDataNodeTaskBuilder { 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 4cca55d705f..5d45383998c 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 @@ public class IoTDBDataNodeReceiver extends IoTDBFileReceiver { 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<String> 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<String> absolutePaths) + private TSStatus loadTsFileAsync( + final String dataBaseName, + final List<String> absolutePaths, + final boolean shouldWaitForSchemaBeforeLoad) throws IOException { final Map<String, String> 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 @@ public class IoTDBDataNodeReceiver extends IoTDBFileReceiver { final boolean shouldConvertDataTypeOnTypeMismatch, final boolean validateTsFile, final boolean shouldMarkAsPipeRequest) { + return buildLoadTsFileAttributesForAsync( + dataBaseName, + shouldConvertDataTypeOnTypeMismatch, + validateTsFile, + shouldMarkAsPipeRequest, + false); + } + + static Map<String, String> 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 @@ public class IoTDBDataNodeReceiver extends IoTDBFileReceiver { 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 25fb874cb6e..70a96572995 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 @@ public class PipeTransferTsFileSealWithModReq extends PipeTransferFileSealReqV2 } 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 class PipeTransferTsFileSealWithModReq extends PipeTransferFileSealReqV2 : 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 class PipeTransferTsFileSealWithModReq extends PipeTransferFileSealReqV2 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 class PipeTransferTsFileSealWithModReq extends PipeTransferFileSealReqV2 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<String, String> generateParameters( + final String tsFileName, + final String dataBaseName, + final boolean shouldWaitForSchemaBeforeLoad) { + final HashMap<String, String> 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 a535f5bb33a..0cd7115bc37 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 @@ public class IoTDBDataRegionAirGapSink extends IoTDBDataNodeAirGapSink { 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 @@ public class IoTDBDataRegionAirGapSink extends IoTDBDataNodeAirGapSink { 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 7f06b771a0c..89791928a01 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 class PipeTransferTsFileHandler extends PipeTransferTrackableHandler { 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 46d4034cf35..25a4f11f16e 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 @@ public class IoTDBDataRegionSyncSink extends IoTDBDataNodeSyncSink { modFile.length(), tsFile.getName(), tsFile.length(), - dataBaseName)); + dataBaseName, + shouldWaitForSchemaBeforeLoad)); pipeName2WeightMap.forEach( (pipePair, weight) -> @@ -583,7 +584,10 @@ public class IoTDBDataRegionSyncSink extends IoTDBDataNodeSyncSink { 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 48d39165840..aa274a100c0 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 class LoadTsFileAnalyzer implements AutoCloseable { 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 class LoadTsFileAnalyzer implements AutoCloseable { 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 @@ public class LoadTsFileAnalyzer implements AutoCloseable { return isAutoCreateDatabase; } + protected boolean isAutoCreateSchema() { + return isAutoCreateSchema; + } + + protected boolean isAutoCreateSchemaAllowed() { + return isAutoCreateSchemaAllowed; + } + protected boolean isConvertOnTypeMismatch() { return isConvertOnTypeMismatch; } @@ -298,6 +316,7 @@ public class LoadTsFileAnalyzer implements AutoCloseable { databaseLevel, isConvertOnTypeMismatch, isVerifySchema, + isAutoCreateSchemaAllowed, tabletConversionThresholdBytes, isGeneratedByPipe, Objects.nonNull(context) ? context.getUsername() : null); @@ -536,8 +555,7 @@ public class LoadTsFileAnalyzer implements AutoCloseable { getOrCreateTreeSchemaVerifier().setCurrentModificationsAndTimeIndex(tsFileResource, reader); - final boolean isAutoCreateSchemaOrVerifySchemaEnabled = - IoTDBDescriptor.getInstance().getConfig().isAutoCreateSchemaEnabled() || isVerifySchema(); + final boolean isAutoCreateSchemaOrVerifySchemaEnabled = isAutoCreateSchema || isVerifySchema(); while (timeseriesMetadataIterator.hasNext()) { final Map<IDeviceID, List<TimeseriesMetadata>> device2TimeseriesMetadata = timeseriesMetadataIterator.next(); @@ -846,9 +864,7 @@ public class LoadTsFileAnalyzer implements AutoCloseable { } 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 6d1704c0cfd..5bb12270e00 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 @@ public class TreeSchemaAutoCreatorAndVerifier { 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 f56c3e2c92f..25dd523f3f0 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.path.PathPatternTree; 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 @@ public interface ISchemaFetcher { List<Boolean> aligned, MPPQueryContext context); + default ISchemaTree fetchSchemaList( + final List<PartialPath> devicePaths, + final List<String[]> 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<Template, PartialPath> checkTemplateSetInfo(PartialPath devicePath); Pair<Template, PartialPath> 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 ff6efc8b3c1..6123afb1d9e 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 class SchemaValidator { List<CompressionType[]> compressionTypes, List<Boolean> 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<PartialPath> devicePaths, + final List<String[]> measurements, + final List<TSDataType[]> dataTypes, + final List<TSEncoding[]> encodings, + final List<CompressionType[]> compressionTypes, + final List<Boolean> 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 28ae4d7f081..17f95824929 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 @@ public class LoadTsFile extends Statement { 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 class LoadTsFile extends Statement { return verify; } + public boolean isAutoCreateSchema() { + return autoCreateSchema; + } + public int getDatabaseLevel() { return databaseLevel; } @@ -225,6 +231,8 @@ public class LoadTsFile extends Statement { 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 class LoadTsFile extends Statement { 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 0d632ef87bf..ca823fcb557 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 java.util.Map; 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 @@ public class LoadTsFileStatement extends Statement { 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 @@ public class LoadTsFileStatement extends Statement { 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 class LoadTsFileStatement extends Statement { 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 @@ public class LoadTsFileStatement extends Statement { 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 class LoadTsFileStatement extends Statement { 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 class LoadTsFileStatement extends Statement { 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 class LoadTsFileStatement extends Statement { + 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 991d368d395..7792cae1a77 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 final class ActiveLoadPathHelper { 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 final class ActiveLoadPathHelper { 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 final class ActiveLoadPathHelper { 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 3076afe1f25..8d4755aaf70 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 class LoadTsFileConfigurator { 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 class LoadTsFileConfigurator { 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<String, String> 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 e00f9500bbe..f3ce153970d 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.apache.iotdb.pipe.api.customizer.parameter.PipeParameters; 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 class PipeDataNodeTaskBuilderTest { Boolean.TRUE.toString(), sinkParameters.getStringByKeys(PipeSinkConstant.CONNECTOR_USE_EVENT_USER_NAME_KEY)); } + + @Test + public void testPreprocessParametersWaitsForCompleteIoTDBSchemaHistory() { + final Map<String, String> 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<String, String> sourceAttributes = new HashMap<>(); + sourceAttributes.put(PipeSourceConstant.SOURCE_INCLUSION_KEY, "all"); + sourceAttributes.put(PipeSourceConstant.SOURCE_EXCLUSION_KEY, excludedOption); + final Map<String, String> 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<String, String> 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<String, String> sourceAttributes = new HashMap<>(); + sourceAttributes.put(PipeSourceConstant.SOURCE_INCLUSION_KEY, "all"); + final Map<String, String> 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<String, String> 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 8f2e86c62d0..79576ed2680 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 class IoTDBDataNodeReceiverTest { 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<String, String> 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 07b7661e2cb..92d3e0e6877 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 class PipeDataNodeThriftRequestTest { 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 class PipeDataNodeThriftRequestTest { 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 68f4bde29e6..cda3515db9e 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 class LoadTsFileAnalyzerTest { } } + @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 60532f749d8..16695553340 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 class ActiveLoadDirScannerTest { // 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<String, String> 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 77936a653c7..c339aef5977 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 class ActiveLoadPathHelperTest { 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 class ActiveLoadPathHelperTest { } } + @Test + public void testAutoCreateSchemaAttributeShouldSurviveActiveLoadPath() throws Exception { + final File pendingDir = Files.createTempDirectory("active-load-schema").toFile(); + try { + final Map<String, String> 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<String, String> 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 152b8c008f3..f1ca66be098 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 9a4f2d09ce7..3bde97c1a12 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 class PipeInclusionOptions { return options; } + public static boolean areOptionsEnabled(final PipeParameters parameters, final String... options) + throws IllegalPathException { + final Set<PartialPath> inclusionOptions = parseOptions(getInclusionString(parameters)); + final Set<PartialPath> 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 9362b62f414..eb2fdfd73d0 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.common.rpc.thrift.TEndPoint; 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 abstract class IoTDBSink implements PipeConnector, PipeConnectorWithEvent 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 abstract class IoTDBSink implements PipeConnector, PipeConnectorWithEvent return receiverStatusHandler; } + public boolean shouldWaitForSchemaBeforeLoad() { + return shouldWaitForSchemaBeforeLoad; + } + public void setTabletBatchSizeHistogram(Histogram tabletBatchSizeHistogram) { // do nothing by default }
