Caideyipi commented on code in PR #17948:
URL: https://github.com/apache/iotdb/pull/17948#discussion_r3489179346
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/task/PipeTaskCoordinator.java:
##########
@@ -113,19 +120,97 @@ public TSStatus createPipe(TCreatePipeReq req) {
return status;
}
+ private boolean isDoubleLivingPipe(final TCreatePipeReq req) {
+ return new PipeParameters(req.getExtractorAttributes())
+ .getBooleanOrDefault(
+ Arrays.asList(
+ PipeSourceConstant.EXTRACTOR_MODE_DOUBLE_LIVING_KEY,
+ PipeSourceConstant.SOURCE_MODE_DOUBLE_LIVING_KEY),
+ PipeSourceConstant.EXTRACTOR_MODE_DOUBLE_LIVING_DEFAULT_VALUE);
+ }
+
+ private TSStatus createDoubleLivingPipe(final TCreatePipeReq req) {
+ final PipeParameters sourceParameters = new
PipeParameters(req.getExtractorAttributes());
+ final String currentDialect =
+ sourceParameters.getStringOrDefault(
+ SystemConstant.SQL_DIALECT_KEY,
SystemConstant.SQL_DIALECT_TREE_VALUE);
+ final String firstDialect =
+ SystemConstant.SQL_DIALECT_TREE_VALUE.equals(currentDialect)
+ ? SystemConstant.SQL_DIALECT_TREE_VALUE
+ : SystemConstant.SQL_DIALECT_TABLE_VALUE;
+ final String secondDialect =
+ SystemConstant.SQL_DIALECT_TREE_VALUE.equals(firstDialect)
+ ? SystemConstant.SQL_DIALECT_TABLE_VALUE
+ : SystemConstant.SQL_DIALECT_TREE_VALUE;
+
+ final TCreatePipeReq firstReq =
+ cloneCreatePipeRequestWithDialect(req, sourceParameters, firstDialect);
+ final TCreatePipeReq secondReq =
+ cloneCreatePipeRequestWithDialect(req, sourceParameters,
secondDialect);
+ try {
+ pipeTaskInfo.checkBeforeCreatePipe(firstReq);
+ pipeTaskInfo.checkBeforeCreatePipe(secondReq);
+ } catch (final Exception e) {
+ return RpcUtils.getStatus(TSStatusCode.PIPE_ERROR, e.getMessage());
+ }
+
+ final TSStatus firstStatus =
configManager.getProcedureManager().createPipe(firstReq);
+ if (TSStatusCode.SUCCESS_STATUS.getStatusCode() != firstStatus.getCode()) {
+ return firstStatus;
+ }
+ final TSStatus secondStatus =
configManager.getProcedureManager().createPipe(secondReq);
+ if (TSStatusCode.SUCCESS_STATUS.getStatusCode() == secondStatus.getCode())
{
+ return secondStatus;
+ }
+ if (!firstReq.isSetIfNotExistsCondition() ||
!firstReq.isIfNotExistsCondition()) {
+ configManager
+ .getProcedureManager()
+ .dropPipe(
+ firstReq.getPipeName(),
SystemConstant.SQL_DIALECT_TABLE_VALUE.equals(firstDialect));
+ }
+ return secondStatus;
+ }
+
+ private TCreatePipeReq cloneCreatePipeRequestWithDialect(
+ final TCreatePipeReq req, final PipeParameters sourceParameters, final
String sqlDialect) {
+ final Map<String, String> sourceAttributes = new
HashMap<>(sourceParameters.getAttribute());
+ sourceAttributes.put(SystemConstant.SQL_DIALECT_KEY, sqlDialect);
+
+ final TCreatePipeReq clonedReq =
+ new TCreatePipeReq()
+ .setPipeName(req.getPipeName())
+ .setExtractorAttributes(sourceAttributes)
+
.setProcessorAttributes(cloneAttributes(req.getProcessorAttributes()))
+
.setConnectorAttributes(cloneAttributes(req.getConnectorAttributes()));
Review Comment:
Kept the copies intentionally. The cloned dialect-specific requests mutate
source attributes by removing double-living flags and setting
dialect/visibility, and validation/procedure code may further normalize mutable
maps. Copying keeps the original request and the two dialect-specific requests
isolated from each other.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
