Caideyipi commented on code in PR #17948:
URL: https://github.com/apache/iotdb/pull/17948#discussion_r3489179011
##########
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));
+ }
Review Comment:
Fixed in e5dc1a6. `createDoubleLivingPipe` now uses the boolean result from
`checkBeforeCreatePipe` for each dialect, so `IF NOT EXISTS` only creates the
missing side. If the second creation fails, the first side is rolled back only
when it was newly created by this call.
--
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]
