This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 71700468942 [Pipe] Fix historical TsFile schema recovery race (#18314)
71700468942 is described below
commit 71700468942206490572314bd32a8f8b65f3032f
Author: Caideyipi <[email protected]>
AuthorDate: Fri Jul 31 10:01:06 2026 +0800
[Pipe] Fix historical TsFile schema recovery race (#18314)
* [Pipe] Fix historical TsFile schema recovery race
* Clarify load schema creation and lock lifecycle
---
.../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 | 25 +++++--
.../load/TreeSchemaAutoCreatorAndVerifier.java | 1 +
.../plan/analyze/schema/ISchemaFetcher.java | 19 ++++++
.../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, 566 insertions(+), 47 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 46cbf4f1b82..7424f20bb8a 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 1da56f5703d..e45cfc89ea9 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 e46b25d8ff2..9ff52f3ba9d 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
@@ -202,9 +202,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 c1a1b9b7e6e..ced4ec12272 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..34827de0c63 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 isAutoCreateSchemaRequested;
+ private final boolean isAutoCreateSchemaEnabled;
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.isAutoCreateSchemaRequested =
loadTsFileStatement.isAutoCreateSchema();
+ this.isAutoCreateSchemaEnabled =
+ IoTDBDescriptor.getInstance().getConfig().isAutoCreateSchemaEnabled()
+ && isAutoCreateSchemaRequested;
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.isAutoCreateSchemaRequested =
loadTsFileTableStatement.isAutoCreateSchema();
+ this.isAutoCreateSchemaEnabled =
+ IoTDBDescriptor.getInstance().getConfig().isAutoCreateSchemaEnabled()
+ && isAutoCreateSchemaRequested;
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 isAutoCreateSchemaEnabled() {
+ return isAutoCreateSchemaEnabled;
+ }
+
+ protected boolean isAutoCreateSchemaRequested() {
+ return isAutoCreateSchemaRequested;
+ }
+
protected boolean isConvertOnTypeMismatch() {
return isConvertOnTypeMismatch;
}
@@ -298,6 +316,7 @@ public class LoadTsFileAnalyzer implements AutoCloseable {
databaseLevel,
isConvertOnTypeMismatch,
isVerifySchema,
+ isAutoCreateSchemaRequested,
tabletConversionThresholdBytes,
isGeneratedByPipe,
Objects.nonNull(context) ? context.getUsername() : null);
@@ -537,7 +556,7 @@ public class LoadTsFileAnalyzer implements AutoCloseable {
getOrCreateTreeSchemaVerifier().setCurrentModificationsAndTimeIndex(tsFileResource,
reader);
final boolean isAutoCreateSchemaOrVerifySchemaEnabled =
- IoTDBDescriptor.getInstance().getConfig().isAutoCreateSchemaEnabled()
|| isVerifySchema();
+ isAutoCreateSchemaEnabled || isVerifySchema();
while (timeseriesMetadataIterator.hasNext()) {
final Map<IDeviceID, List<TimeseriesMetadata>> device2TimeseriesMetadata
=
timeseriesMetadataIterator.next();
@@ -846,9 +865,7 @@ public class LoadTsFileAnalyzer implements AutoCloseable {
}
boolean isTemporaryUnavailableDueToPipeSchemaNotReady(final Throwable
throwable) {
- if (!isGeneratedByPipe
- || !isVerifySchema
- ||
IoTDBDescriptor.getInstance().getConfig().isAutoCreateSchemaEnabled()) {
+ 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 3e968073cc0..d92c5954668 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
@@ -519,6 +519,7 @@ public class TreeSchemaAutoCreatorAndVerifier {
encodingsList,
compressionTypesList,
isAlignedList,
+ 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 f56c3e2c92f..7cf73c942ad 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,23 @@ public interface ISchemaFetcher {
List<Boolean> aligned,
MPPQueryContext context);
+ default ISchemaTree fetchSchemaList(
+ final List<PartialPath> devicePaths,
+ final List<String[]> 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();
+ 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 59b247d4d34..a0999d9f0a6 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 1a11de1eab5..68503f3d794 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
@@ -299,6 +299,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.isAutoCreateSchemaRequested());
+ 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.isAutoCreateSchemaRequested());
+ 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.isAutoCreateSchemaEnabled());
+ Assert.assertTrue(analyzer.isAutoCreateSchemaRequested());
+ } 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
}
