This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch dev/1.3
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/dev/1.3 by this push:
new 85f82fd1030 [Pipe] Fix historical TsFile schema recovery race (#18314)
(#18378)
85f82fd1030 is described below
commit 85f82fd10309782c6eefe9782d5b5853231cd19a
Author: Caideyipi <[email protected]>
AuthorDate: Mon Aug 3 10:23:03 2026 +0800
[Pipe] Fix historical TsFile schema recovery race (#18314) (#18378)
* [Pipe] Fix historical TsFile schema recovery race
* Clarify load schema creation and lock lifecycle
---
.../it/autocreate/IoTDBPipeAutoConflictIT.java | 5 ++
.../task/builder/PipeDataNodeTaskBuilder.java | 56 +++++++++++++--
.../protocol/thrift/IoTDBDataNodeReceiver.java | 72 ++++++++++++++++----
.../request/PipeTransferTsFileSealWithModReq.java | 77 ++++++++++++++++++---
.../protocol/airgap/IoTDBDataRegionAirGapSink.java | 11 ++-
.../async/handler/PipeTransferTsFileHandler.java | 10 ++-
.../thrift/sync/IoTDBDataRegionSyncSink.java | 10 ++-
.../plan/analyze/load/LoadTsFileAnalyzer.java | 22 ++++--
.../plan/analyze/schema/ISchemaFetcher.java | 19 ++++++
.../plan/analyze/schema/SchemaValidator.java | 34 +++++++++-
.../plan/statement/crud/LoadTsFileStatement.java | 16 +++++
.../load/active/ActiveLoadPathHelper.java | 15 ++++
.../load/config/LoadTsFileConfigurator.java | 22 ++++++
.../task/builder/PipeDataNodeTaskBuilderTest.java | 79 ++++++++++++++++++++++
.../protocol/thrift/IoTDBDataNodeReceiverTest.java | 30 ++++++++
.../pipe/sink/PipeDataNodeThriftRequestTest.java | 30 ++++++++
.../plan/analyze/load/LoadTsFileAnalyzerTest.java | 72 +++++++++++++++++++-
.../load/active/ActiveLoadPathHelperTest.java | 73 ++++++++++++++++++++
.../pipe/config/constant/SystemConstant.java | 2 +
.../options/PipeInclusionOptions.java | 31 +++++++++
.../commons/pipe/sink/protocol/IoTDBSink.java | 8 +++
21 files changed, 644 insertions(+), 50 deletions(-)
diff --git
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeAutoConflictIT.java
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeAutoConflictIT.java
index 3b5d94f4654..9c678915e32 100644
---
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeAutoConflictIT.java
+++
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeAutoConflictIT.java
@@ -392,6 +392,11 @@ public class IoTDBPipeAutoConflictIT extends
AbstractPipeDualAutoIT {
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", "count(devices),",
Collections.singleton("3,"));
}
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 7ea16f888a1..e6d3dee4167 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
@@ -56,11 +56,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;
@@ -261,13 +270,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(
+ "PipeDataNodeTaskBuilder failed to parse 'inclusion' and 'exclusion'
parameters: {}",
+ 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 6ad15376f85..92ff5ca2de5 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
@@ -503,32 +503,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 {
- if (req instanceof PipeTransferTsFileSealWithModReq) {
- final String dataBaseName =
- ((PipeTransferTsFileSealWithModReq)
req).getDatabaseNameByTsFileName();
- return isUsingAsyncLoadTsFileStrategy.get()
- ? loadTsFileAsync(dataBaseName, fileAbsolutePaths)
- : loadTsFileSync(dataBaseName,
fileAbsolutePaths.get(req.getFileNames().size() - 1));
+ if (!(req instanceof PipeTransferTsFileSealWithModReq)) {
+ return loadSchemaSnapShot(req.getParameters(), fileAbsolutePaths);
}
- 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 (!ActiveLoadUtil.loadFilesToActiveDir(loadAttributes, absolutePaths,
true)) {
throw new PipeException("Load active listening pipe dir is not set.");
}
@@ -540,23 +550,42 @@ 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);
}
- 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(
@@ -565,10 +594,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 7d0aa99cb13..63b3dbfd296 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
@@ -41,6 +41,7 @@ public class PipeTransferTsFileSealWithModReq extends
PipeTransferFileSealReqV2
}
private 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 getParameters() == null
@@ -50,15 +51,27 @@ public class PipeTransferTsFileSealWithModReq extends
PipeTransferFileSealReqV2
generateDatabaseNameWithFileNameKey(getFileNames().get(getFileNames().size() -
1)));
}
+ public boolean shouldWaitForSchemaBeforeLoad() {
+ return getParameters() != null
+ &&
Boolean.parseBoolean(getParameters().get(WAIT_FOR_SCHEMA_BEFORE_LOAD_KEY));
+ }
+
private static String generateDatabaseNameWithFileNameKey(final String
fileName) {
return DATABASE_NAME_KEY_PREFIX + fileName;
}
- private static Map<String, String> generateDatabaseNameParameter(
- final String tsFileName, final String dataBaseName) {
- return dataBaseName == null
- ? new HashMap<>()
- :
Collections.singletonMap(generateDatabaseNameWithFileNameKey(tsFileName),
dataBaseName);
+ private static Map<String, String> generateParameters(
+ final String tsFileName,
+ final String dataBaseName,
+ final boolean shouldWaitForSchemaBeforeLoad) {
+ final Map<String, String> parameters = new HashMap<>();
+ if (dataBaseName != null) {
+ parameters.put(generateDatabaseNameWithFileNameKey(tsFileName),
dataBaseName);
+ }
+ if (shouldWaitForSchemaBeforeLoad) {
+ parameters.put(WAIT_FOR_SCHEMA_BEFORE_LOAD_KEY, Boolean.TRUE.toString());
+ }
+ return parameters;
}
/////////////////////////////// Thrift ///////////////////////////////
@@ -66,7 +79,7 @@ public class PipeTransferTsFileSealWithModReq extends
PipeTransferFileSealReqV2
public static PipeTransferTsFileSealWithModReq toTPipeTransferReq(
String modFileName, long modFileLength, String tsFileName, long
tsFileLength)
throws IOException {
- return toTPipeTransferReq(modFileName, modFileLength, tsFileName,
tsFileLength, null);
+ return toTPipeTransferReq(modFileName, modFileLength, tsFileName,
tsFileLength, null, false);
}
public static PipeTransferTsFileSealWithModReq toTPipeTransferReq(
@@ -76,23 +89,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),
- generateDatabaseNameParameter(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),
- generateDatabaseNameParameter(tsFileName, dataBaseName));
+ generateParameters(tsFileName, dataBaseName,
shouldWaitForSchemaBeforeLoad));
}
public static PipeTransferTsFileSealWithModReq
fromTPipeTransferReq(TPipeTransferReq req) {
@@ -105,7 +139,7 @@ public class PipeTransferTsFileSealWithModReq extends
PipeTransferFileSealReqV2
public static byte[] toTPipeTransferBytes(
String modFileName, long modFileLength, String tsFileName, long
tsFileLength)
throws IOException {
- return toTPipeTransferBytes(modFileName, modFileLength, tsFileName,
tsFileLength, null);
+ return toTPipeTransferBytes(modFileName, modFileLength, tsFileName,
tsFileLength, null, false);
}
public static byte[] toTPipeTransferBytes(
@@ -115,21 +149,42 @@ 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),
- generateDatabaseNameParameter(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),
- generateDatabaseNameParameter(tsFileName, dataBaseName));
+ generateParameters(tsFileName, dataBaseName,
shouldWaitForSchemaBeforeLoad));
}
/////////////////////////////// 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 4b088b0aefd..27564aa1505 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
@@ -406,7 +406,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())
@@ -420,10 +425,10 @@ public class IoTDBDataRegionAirGapSink extends
IoTDBDataNodeAirGapSink {
transferFilePieces(pipe2WeightMap, tsFile, socket, false);
if (!sendWeighted(
socket,
- dataBaseName == null
+ dataBaseName == null && !shouldWaitForSchemaBeforeLoad
?
PipeTransferTsFileSealReq.toTPipeTransferBytes(tsFile.getName(),
tsFile.length())
: 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 fc6dfefc24f..d5a10adf75e 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
@@ -196,12 +196,16 @@ public class PipeTransferTsFileHandler extends
PipeTransferTrackableHandler {
modFile.length(),
tsFile.getName(),
tsFile.length(),
- dataBaseName)
- : dataBaseName == null
+ dataBaseName,
+ sink.shouldWaitForSchemaBeforeLoad())
+ : dataBaseName == null && !sink.shouldWaitForSchemaBeforeLoad()
? PipeTransferTsFileSealReq.toTPipeTransferReq(
tsFile.getName(), tsFile.length())
: 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 49d52fac3a7..6b388a6b2be 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
@@ -470,7 +470,8 @@ public class IoTDBDataRegionSyncSink extends
IoTDBDataNodeSyncSink {
modFile.length(),
tsFile.getName(),
tsFile.length(),
- dataBaseName));
+ dataBaseName,
+ shouldWaitForSchemaBeforeLoad));
pipeName2WeightMap.forEach(
(pipePair, weight) ->
@@ -495,11 +496,14 @@ public class IoTDBDataRegionSyncSink extends
IoTDBDataNodeSyncSink {
try {
final TPipeTransferReq req =
compressIfNeeded(
- dataBaseName == null
+ dataBaseName == null && !shouldWaitForSchemaBeforeLoad
? PipeTransferTsFileSealReq.toTPipeTransferReq(
tsFile.getName(), tsFile.length())
: 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 596086cb2b3..227064b286a 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
@@ -154,6 +154,8 @@ public class LoadTsFileAnalyzer implements AutoCloseable {
private final int databaseLevel;
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;
@@ -180,12 +182,24 @@ public class LoadTsFileAnalyzer implements AutoCloseable {
this.databaseLevel = loadTsFileStatement.getDatabaseLevel();
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();
this.tabletConversionThresholdBytes =
loadTsFileStatement.getTabletConversionThresholdBytes();
}
+ protected boolean isAutoCreateSchemaEnabled() {
+ return isAutoCreateSchemaEnabled;
+ }
+
+ protected boolean isAutoCreateSchemaRequested() {
+ return isAutoCreateSchemaRequested;
+ }
+
public Analysis analyzeFileByFile(Analysis analysis) {
if (!checkBeforeAnalyzeFileByFile(analysis)) {
return analysis;
@@ -255,6 +269,7 @@ public class LoadTsFileAnalyzer implements AutoCloseable {
databaseLevel,
isConvertOnTypeMismatch,
isVerifySchema,
+ isAutoCreateSchemaRequested,
tabletConversionThresholdBytes,
isGeneratedByPipe);
if (ActiveLoadUtil.loadTsFileAsyncToActiveDir(
@@ -435,7 +450,7 @@ public class LoadTsFileAnalyzer implements AutoCloseable {
schemaAutoCreatorAndVerifier.setCurrentModificationsAndTimeIndex(tsFileResource);
final boolean isAutoCreateSchemaOrVerifySchemaEnabled =
- IoTDBDescriptor.getInstance().getConfig().isAutoCreateSchemaEnabled()
|| isVerifySchema;
+ isAutoCreateSchemaEnabled || isVerifySchema;
while (timeseriesMetadataIterator.hasNext()) {
final Map<IDeviceID, List<TimeseriesMetadata>> device2TimeseriesMetadata
=
@@ -574,9 +589,7 @@ public class LoadTsFileAnalyzer implements AutoCloseable {
}
boolean isTemporaryUnavailableDueToPipeSchemaNotReady(final Throwable
throwable) {
- if (!isGeneratedByPipe
- || !isVerifySchema
- ||
IoTDBDescriptor.getInstance().getConfig().isAutoCreateSchemaEnabled()) {
+ if (!isGeneratedByPipe || !isVerifySchema || isAutoCreateSchemaEnabled) {
return false;
}
@@ -961,6 +974,7 @@ public class LoadTsFileAnalyzer implements AutoCloseable {
encodingsList,
compressionTypesList,
isAlignedList,
+ isAutoCreateSchemaRequested,
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 56a7670ad9d..0bb8ab80706 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
@@ -23,6 +23,8 @@ import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.commons.path.PathPatternTree;
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.iotdb.db.schemaengine.template.Template;
import org.apache.tsfile.enums.TSDataType;
@@ -109,6 +111,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);
+ 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);
+ }
+
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 7bbdd5cefc6..443d9351cae 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
@@ -64,7 +64,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/statement/crud/LoadTsFileStatement.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/crud/LoadTsFileStatement.java
index 404b957c786..ae57cfacc25 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
@@ -52,6 +52,7 @@ public class LoadTsFileStatement extends Statement {
private int databaseLevel;
private String database;
private boolean verifySchema = true;
+ private boolean autoCreateSchema = true;
private boolean deleteAfterLoad = false;
private boolean convertOnTypeMismatch = true;
private long tabletConversionThresholdBytes = -1;
@@ -76,6 +77,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 =
@@ -118,6 +120,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.autoCreateDatabase =
IoTDBDescriptor.getInstance().getConfig().isAutoCreateSchemaEnabled();
@@ -222,6 +225,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;
@@ -301,6 +312,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();
@@ -390,6 +403,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;
@@ -436,6 +450,8 @@ public class LoadTsFileStatement extends Statement {
+ database
+ ", 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 2503b822b9e..c9e33ffb5c2 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
@@ -51,6 +51,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.PIPE_GENERATED_KEY));
private ActiveLoadPathHelper() {
@@ -61,6 +62,7 @@ public final class ActiveLoadPathHelper {
final Integer databaseLevel,
final Boolean convertOnTypeMismatch,
final Boolean verify,
+ final Boolean autoCreateSchema,
final Long tabletConversionThresholdBytes,
final Boolean pipeGenerated) {
return buildAttributes(
@@ -68,6 +70,7 @@ public final class ActiveLoadPathHelper {
databaseLevel,
convertOnTypeMismatch,
verify,
+ autoCreateSchema,
tabletConversionThresholdBytes,
pipeGenerated);
}
@@ -77,6 +80,7 @@ public final class ActiveLoadPathHelper {
final Integer databaseLevel,
final Boolean convertOnTypeMismatch,
final Boolean verify,
+ final Boolean autoCreateSchema,
final Long tabletConversionThresholdBytes,
final Boolean pipeGenerated) {
final Map<String, String> attributes = new LinkedHashMap<>();
@@ -105,6 +109,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());
}
@@ -204,6 +213,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();
@@ -258,6 +270,9 @@ public final class ActiveLoadPathHelper {
case LoadTsFileConfigurator.VERIFY_KEY:
LoadTsFileConfigurator.validateVerifyParam(value);
break;
+ case LoadTsFileConfigurator.AUTO_CREATE_SCHEMA_KEY:
+ LoadTsFileConfigurator.validateAutoCreateSchemaParam(value);
+ break;
default:
LoadTsFileConfigurator.validateParameters(key, value);
}
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 8b689c6fb22..7ae75a6ca68 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
@@ -49,6 +49,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;
@@ -165,6 +168,25 @@ 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(
+ "Given %s value '%s' is not supported, please input a valid
boolean value.",
+ 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 a2710535f70..82dea37f52e 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
@@ -68,6 +68,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 95f1cdd146c..3bdadc04cf4 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
@@ -35,6 +35,7 @@ import
org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferTable
import
org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferTsFilePieceReq;
import
org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferTsFilePieceWithModReq;
import
org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferTsFileSealReq;
+import
org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferTsFileSealWithModReq;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeId;
import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.metedata.write.CreateAlignedTimeSeriesNode;
import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowNode;
@@ -59,6 +60,7 @@ import java.io.IOException;
import java.nio.ByteBuffer;
import java.time.LocalDate;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collections;
import java.util.List;
@@ -492,6 +494,34 @@ public class PipeDataNodeThriftRequestTest {
Assert.assertEquals(req.getFileLength(), deserializeReq.getFileLength());
}
+ @Test
+ public void testPipeTransferTsFileSealWithModReq() throws IOException {
+ final String modFileName = "1.tsfile.mod";
+ final String tsFileName = "1.tsfile";
+
+ final PipeTransferTsFileSealWithModReq req =
+ PipeTransferTsFileSealWithModReq.toTPipeTransferReq(
+ modFileName, 10, tsFileName, 100, "root.db");
+ final PipeTransferTsFileSealWithModReq deserializeReq =
+ PipeTransferTsFileSealWithModReq.fromTPipeTransferReq(req);
+
+ 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
public void testPipeTransferSchemaSnapshotSealReq() throws IOException {
final String mTreeSnapshotName = "mtree.snapshot";
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 bea5e0aa981..589542c8023 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
@@ -20,7 +20,9 @@
package org.apache.iotdb.db.queryengine.plan.analyze.load;
import org.apache.iotdb.commons.path.PartialPath;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.exception.LoadAnalyzeException;
+import org.apache.iotdb.db.exception.LoadAnalyzeMissingSchemaException;
import org.apache.iotdb.db.queryengine.common.MPPQueryContext;
import org.apache.iotdb.db.queryengine.common.QueryId;
import org.apache.iotdb.db.queryengine.plan.analyze.IPartitionFetcher;
@@ -100,10 +102,76 @@ 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);
+ waitingStatement.markIsGeneratedByPipe();
+ try (final LoadTsFileAnalyzer waitingAnalyzer =
+ createAnalyzer(waitingStatement, "load_pipe_waiting_test")) {
+ Assert.assertFalse(waitingAnalyzer.isAutoCreateSchemaRequested());
+ Assert.assertTrue(
+ waitingAnalyzer.isTemporaryUnavailableDueToPipeSchemaNotReady(
+ new LoadAnalyzeMissingSchemaException("missing schema")));
+ }
+
+ final LoadTsFileStatement defaultStatement =
+ LoadTsFileStatement.createUnchecked(tsFile.getAbsolutePath());
+ defaultStatement.markIsGeneratedByPipe();
+ try (final LoadTsFileAnalyzer defaultAnalyzer =
+ createAnalyzer(defaultStatement, "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 =
+ createAnalyzer(
+ LoadTsFileStatement.createUnchecked(tsFile.getAbsolutePath()),
+ "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 LoadTsFileAnalyzer createAnalyzer(final File tsFile) throws
Exception {
+ return createAnalyzer(
+ LoadTsFileStatement.createUnchecked(tsFile.getAbsolutePath()),
"load_tree_test");
+ }
+
+ private LoadTsFileAnalyzer createAnalyzer(
+ final LoadTsFileStatement statement, final String queryId) throws
Exception {
return new LoadTsFileAnalyzer(
- LoadTsFileStatement.createUnchecked(tsFile.getAbsolutePath()),
- new MPPQueryContext(new QueryId("load_tree_test")),
+ statement,
+ new MPPQueryContext(new QueryId(queryId)),
mock(IPartitionFetcher.class),
mock(ISchemaFetcher.class));
}
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
new file mode 100644
index 00000000000..1a75dcce142
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadPathHelperTest.java
@@ -0,0 +1,73 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.storageengine.load.active;
+
+import org.apache.iotdb.db.queryengine.plan.statement.crud.LoadTsFileStatement;
+import org.apache.iotdb.db.storageengine.load.config.LoadTsFileConfigurator;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.io.File;
+import java.nio.file.Files;
+import java.util.Map;
+
+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);
+ final File targetDir = ActiveLoadPathHelper.resolveTargetDir(pendingDir,
attributes);
+ Assert.assertTrue(targetDir.mkdirs());
+ final File tsFile = new File(targetDir, "1-0-0-0.tsfile");
+ Assert.assertTrue(tsFile.createNewFile());
+
+ 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;
+ }
+ final File[] children = file.listFiles();
+ if (children != null) {
+ for (final File child : children) {
+ deleteRecursively(child);
+ }
+ }
+ Assert.assertTrue(file.delete());
+ }
+}
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 a4119979e76..f20150a96ce 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
@@ -24,6 +24,8 @@ public class SystemConstant {
// This can be arbitrarily changed since it's only a memory key and not
stored
public static final String RESTART_OR_NEWLY_ADDED_KEY =
"__system.restart_or_newly_added";
public static final boolean RESTART_OR_NEWLY_ADDED_DEFAULT_VALUE = false;
+ public static final String SINK_WAIT_FOR_SCHEMA_BEFORE_LOAD_KEY =
+ "__system.sink-wait-for-schema-before-load";
private SystemConstant() {
throw new IllegalStateException("Utility class");
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 cdcb8866289..280877e3bba 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
@@ -21,6 +21,8 @@ package org.apache.iotdb.commons.pipe.datastructure.options;
import org.apache.iotdb.commons.exception.IllegalPathException;
import org.apache.iotdb.commons.path.PartialPath;
+import org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant;
+import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -192,6 +194,35 @@ public class PipeInclusionOptions {
return options;
}
+ public static String getInclusionString(final PipeParameters parameters) {
+ return parameters.getStringOrDefault(
+ Arrays.asList(
+ PipeSourceConstant.EXTRACTOR_INCLUSION_KEY,
PipeSourceConstant.SOURCE_INCLUSION_KEY),
+ PipeSourceConstant.EXTRACTOR_INCLUSION_DEFAULT_VALUE);
+ }
+
+ public static String getExclusionString(final PipeParameters parameters) {
+ return parameters.getStringOrDefault(
+ Arrays.asList(
+ PipeSourceConstant.EXTRACTOR_EXCLUSION_KEY,
PipeSourceConstant.SOURCE_EXCLUSION_KEY),
+ PipeSourceConstant.EXTRACTOR_EXCLUSION_DEFAULT_VALUE);
+ }
+
+ 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 71ee0a41a91..1d929411929 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
@@ -21,6 +21,7 @@ package org.apache.iotdb.commons.pipe.sink.protocol;
import org.apache.iotdb.common.rpc.thrift.TEndPoint;
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;
@@ -157,6 +158,7 @@ public abstract class IoTDBSink implements PipeConnector,
PipeConnectorWithEvent
protected String loadTsFileStrategy;
protected boolean loadTsFileValidation;
+ protected boolean shouldWaitForSchemaBeforeLoad;
protected boolean shouldMarkAsPipeRequest;
@@ -282,6 +284,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(
@@ -630,6 +634,10 @@ public abstract class IoTDBSink implements PipeConnector,
PipeConnectorWithEvent
return receiverStatusHandler;
}
+ public boolean shouldWaitForSchemaBeforeLoad() {
+ return shouldWaitForSchemaBeforeLoad;
+ }
+
public void setTabletBatchSizeHistogram(Histogram tabletBatchSizeHistogram) {
// do nothing by default
}
