jt2594838 commented on code in PR #17948:
URL: https://github.com/apache/iotdb/pull/17948#discussion_r3473183976
##########
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:
Necessary to copy them? Will they be modified?
##########
iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/meta/PipeMetaKeeper.java:
##########
@@ -188,4 +241,98 @@ public long exceptionStoppedPipeCount() {
&&
pipeMeta.getRuntimeMeta().getIsStoppedByRuntimeException())
.count();
}
+
+ ///////////////////////////////// Tree & Table Isolation
/////////////////////////////////
+
+ private Optional<PipeMeta> getPipeMetaOptional(String pipeName) {
+ return pipeMetaKeeperKeyOptional(pipeName).map(pipeNameToPipeMetaMap::get);
+ }
+
+ private Optional<PipeMeta> getPipeMetaOptional(String pipeName, boolean
isTableModel) {
+ return pipeMetaKeeperKeyOptional(pipeName,
isTableModel).map(pipeNameToPipeMetaMap::get);
+ }
+
+ private Optional<PipeMeta> getPipeMetaOptional(String pipeName, long
creationTime) {
+ final PipeMeta pipeMeta = pipeNameToPipeMetaMap.get(pipeName);
+ if (Objects.nonNull(pipeMeta) &&
pipeMeta.getStaticMeta().getCreationTime() == creationTime) {
+ return Optional.of(pipeMeta);
+ }
+
+ return pipeNameToPipeMetaMap.values().stream()
+ .filter(
+ meta ->
+ meta.getStaticMeta().getPipeName().equals(pipeName)
+ && meta.getStaticMeta().getCreationTime() == creationTime)
+ .findFirst();
+ }
+
+ private Optional<String> pipeMetaKeeperKeyOptional(String pipeName, boolean
isTableModel) {
+ final String modelKey = generatePipeMetaKeeperKey(pipeName, isTableModel);
+ if (pipeNameToPipeMetaMap.containsKey(modelKey)) {
+ return Optional.of(modelKey);
+ }
+
+ final String bothModelKey =
generatePipeMetaKeeperKeyForVisibility(pipeName, Visibility.BOTH);
+ return pipeNameToPipeMetaMap.containsKey(bothModelKey)
+ ? Optional.of(bothModelKey)
+ : Optional.empty();
+ }
+
+ private Optional<String> pipeMetaKeeperKeyOptional(String pipeName) {
+ final String treeModelKey = generatePipeMetaKeeperKey(pipeName, false);
+ if (pipeNameToPipeMetaMap.containsKey(treeModelKey)) {
+ return Optional.of(treeModelKey);
+ }
+
+ final String bothModelKey =
generatePipeMetaKeeperKeyForVisibility(pipeName, Visibility.BOTH);
+ if (pipeNameToPipeMetaMap.containsKey(bothModelKey)) {
+ return Optional.of(bothModelKey);
+ }
+
+ final String tableModelKey = generatePipeMetaKeeperKey(pipeName, true);
+ if (pipeNameToPipeMetaMap.containsKey(tableModelKey)) {
+ return Optional.of(tableModelKey);
+ }
+
+ final String noneModelKey =
generatePipeMetaKeeperKeyForVisibility(pipeName, Visibility.NONE);
+ return pipeNameToPipeMetaMap.containsKey(noneModelKey)
+ ? Optional.of(noneModelKey)
+ : Optional.empty();
+ }
+
+ private static String generatePipeMetaKeeperKey(final PipeStaticMeta
pipeStaticMeta) {
+ final String pipeName = pipeStaticMeta.getPipeName();
+ final Visibility visibility =
+
VisibilityUtils.calculateFromExtractorParameters(pipeStaticMeta.getSourceParameters());
+ if (Objects.equals(Visibility.TREE_ONLY, visibility)) {
+ return generatePipeMetaKeeperKey(pipeName, false);
+ }
+ if (Objects.equals(Visibility.TABLE_ONLY, visibility)) {
+ return generatePipeMetaKeeperKey(pipeName, true);
+ }
+ if (Objects.equals(Visibility.NONE, visibility)) {
+ return generatePipeMetaKeeperKeyForVisibility(pipeName, Visibility.NONE);
+ }
+ return generatePipeMetaKeeperKeyForVisibility(pipeName, Visibility.BOTH);
+ }
+
+ private static String generatePipeMetaKeeperKey(
+ final String pipeName, final boolean isTableModel) {
+ return (isTableModel ? "table:" : "tree:") + pipeName.length() + ":" +
pipeName;
+ }
+
+ private static String generatePipeMetaKeeperKeyForVisibility(
+ final String pipeName, final Visibility visibility) {
+ return visibility.name().toLowerCase() + ":" + pipeName.length() + ":" +
pipeName;
+ }
Review Comment:
Why add the length?
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfo.java:
##########
@@ -305,6 +368,24 @@ private void checkBeforeStartPipeInternal(final String
pipeName) throws PipeExce
}
}
+ private void checkBeforeStartPipeInternal(final String pipeName, final
boolean isTableModel)
+ throws PipeException {
+ if (!isPipeExisted(pipeName, isTableModel)) {
+ final String exceptionMessage =
+ String.format("Failed to start pipe %s, %s", pipeName,
PIPE_NOT_EXIST_MSG);
+ LOGGER.warn(exceptionMessage);
Review Comment:
i18n
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfo.java:
##########
@@ -597,6 +736,39 @@ public PipeMeta getPipeMetaByPipeName(final String
pipeName) {
}
}
+ public PipeMeta getPipeMetaByPipeName(final String pipeName, final boolean
isTableModel) {
+ acquireReadLock();
+ try {
+ return pipeMetaKeeper.getPipeMetaByPipeName(pipeName, isTableModel);
+ } finally {
+ releaseReadLock();
+ }
+ }
+
+ public PipeMeta getPipeMetaByPipeStaticMeta(final PipeStaticMeta
pipeStaticMeta) {
+ acquireReadLock();
+ try {
+ return pipeMetaKeeper.getPipeMeta(pipeStaticMeta);
+ } finally {
+ releaseReadLock();
+ }
+ }
+
+ public long generateUniqueCreationTime(final String pipeName) {
+ acquireReadLock();
+ try {
+ long creationTime = System.currentTimeMillis();
+ for (final PipeMeta pipeMeta : pipeMetaKeeper.getPipeMetaList()) {
+ if (pipeMeta.getStaticMeta().getPipeName().equals(pipeName)) {
+ creationTime = Math.max(creationTime,
pipeMeta.getStaticMeta().getCreationTime() + 1);
+ }
+ }
+ return creationTime;
+ } finally {
+ releaseReadLock();
+ }
+ }
Review Comment:
Why is a unique creation time needed?
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfo.java:
##########
@@ -93,6 +96,20 @@ public class PipeTaskInfo implements SnapshotProcessor {
private static final Logger LOGGER =
LoggerFactory.getLogger(PipeTaskInfo.class);
private static final String SNAPSHOT_FILE_NAME = "pipe_task_info.bin";
+ private static OptionalLong
tryGetPushPipeMetaRespExceptionMessageCreationTime(
+ final TPushPipeMetaRespExceptionMessage exceptionMessage) {
+ try {
+ if (!Boolean.TRUE.equals(
+
exceptionMessage.getClass().getMethod("isSetCreationTime").invoke(exceptionMessage)))
{
+ return OptionalLong.empty();
+ }
+ return OptionalLong.of(
+ (long)
exceptionMessage.getClass().getMethod("getCreationTime").invoke(exceptionMessage));
+ } catch (final Exception ignored) {
+ return OptionalLong.empty();
+ }
+ }
Review Comment:
Why use reflection?
##########
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:
What if `ifNotExists = true`, but the first pipe is created, and the second
one is not?
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java:
##########
@@ -2219,125 +2241,172 @@ public SettableFuture<ConfigTaskResult> createPipe(
// Syntactic sugar: if full-sync mode is detected (i.e. not snapshot mode,
or both realtime
// and history are true), the pipe is split into history-only and
realtime–only modes.
- final PipeParameters sourcePipeParameters =
- new PipeParameters(createPipeStatement.getSourceAttributes());
- final PipeParameters sinkPipeParameters =
- new PipeParameters(createPipeStatement.getSinkAttributes());
try (final ConfigNodeClient configNodeClient =
CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.CONFIG_REGION_ID)) {
- if (PipeConfig.getInstance().getPipeAutoSplitFullEnabled()
- && PipeDataNodeAgent.task().isFullSync(sourcePipeParameters)) {
- // 1. Send request to create the real-time data synchronization
pipeline
- final TCreatePipeReq realtimeReq =
- new TCreatePipeReq()
- // Append suffix to the pipeline name for real-time data
- .setPipeName(pipeName + "_realtime")
- // NOTE: set if not exists always to true to handle partial
failure
- .setIfNotExistsCondition(true)
- // Use extractor parameters for real-time data
- .setExtractorAttributes(
- sourcePipeParameters
- .addOrReplaceEquivalentAttributesWithClone(
- new PipeParameters(
- ImmutableMap.of(
-
PipeSourceConstant.EXTRACTOR_REALTIME_ENABLE_KEY,
- Boolean.toString(true),
-
PipeSourceConstant.EXTRACTOR_HISTORY_ENABLE_KEY,
- Boolean.toString(false))))
- .getAttribute())
-
.setProcessorAttributes(createPipeStatement.getProcessorAttributes())
-
.setConnectorAttributes(createPipeStatement.getSinkAttributes());
-
- final TSStatus realtimeTsStatus =
configNodeClient.createPipe(realtimeReq);
- // If creation fails, immediately return with exception
- // If the procedure is still running, it's probably stuck on DataNode
- // The pipe creation can ignore this situation and succeed, thus we do
not need to skip in
- // this case
- if (TSStatusCode.SUCCESS_STATUS.getStatusCode() !=
realtimeTsStatus.getCode()
- && TSStatusCode.OVERLAP_WITH_EXISTING_TASK.getStatusCode()
- != realtimeTsStatus.getCode()) {
- future.setException(new IoTDBException(realtimeTsStatus));
- return future;
- }
-
- // 2. Send request to create the historical data synchronization
pipeline
- final Map<String, String> historySinkAttributes =
- sinkPipeParameters.hasAnyAttributes(
- PipeSinkConstant.SINK_ENABLE_SEND_TSFILE_LIMIT,
- PipeSinkConstant.CONNECTOR_ENABLE_SEND_TSFILE_LIMIT)
- ? createPipeStatement.getSinkAttributes()
- : sinkPipeParameters
- .addOrReplaceEquivalentAttributesWithClone(
- new PipeParameters(
- Collections.singletonMap(
- PipeSinkConstant.SINK_ENABLE_SEND_TSFILE_LIMIT,
- Boolean.TRUE.toString())))
- .getAttribute();
-
- final TCreatePipeReq historyReq =
- new TCreatePipeReq()
- // Append suffix to the pipeline name for historical data
- .setPipeName(pipeName + "_history")
-
.setIfNotExistsCondition(createPipeStatement.hasIfNotExistsCondition())
- // Use source parameters for historical data
- .setExtractorAttributes(
- sourcePipeParameters
- .addOrReplaceEquivalentAttributesWithClone(
- new PipeParameters(
- ImmutableMap.of(
-
PipeSourceConstant.EXTRACTOR_REALTIME_ENABLE_KEY,
- Boolean.toString(false),
-
PipeSourceConstant.EXTRACTOR_HISTORY_ENABLE_KEY,
- Boolean.toString(true),
- PipeSourceConstant.EXTRACTOR_MODE_KEY,
-
PipeSourceConstant.EXTRACTOR_MODE_SNAPSHOT_VALUE,
- // We force the historical pipe to
transfer data (and maybe
- // deletion) only
- // Thus we can transfer schema only once
- // And may drop the historical pipe on
successfully transferred
- PipeSourceConstant.SOURCE_INCLUSION_KEY,
- DataRegionListeningFilter
-
.parseInsertionDeletionListeningOptionPair(
- sourcePipeParameters)
- .getRight()
- ? "data"
- :
PipeSourceConstant.EXTRACTOR_INCLUSION_DEFAULT_VALUE,
- PipeSourceConstant.SOURCE_EXCLUSION_KEY,
-
PipeSourceConstant.EXTRACTOR_EXCLUSION_DEFAULT_VALUE)))
- .getAttribute())
-
.setProcessorAttributes(createPipeStatement.getProcessorAttributes())
- .setConnectorAttributes(historySinkAttributes);
-
- final TSStatus historyTsStatus =
configNodeClient.createPipe(historyReq);
- // If creation fails, immediately return with exception
- if (TSStatusCode.SUCCESS_STATUS.getStatusCode() !=
historyTsStatus.getCode()) {
- future.setException(new IoTDBException(historyTsStatus));
- return future;
- }
-
- // 3. Set success status only if both pipelines are created
successfully
- future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
+ final TSStatus tsStatus =
+ createPipeInternal(
+ configNodeClient, createPipeStatement, sourcePipeParameters,
sinkPipeParameters);
+ if (TSStatusCode.SUCCESS_STATUS.getStatusCode() != tsStatus.getCode()) {
+ future.setException(new IoTDBException(tsStatus));
} else {
- final TCreatePipeReq req =
- new TCreatePipeReq()
- .setPipeName(pipeName)
-
.setIfNotExistsCondition(createPipeStatement.hasIfNotExistsCondition())
-
.setExtractorAttributes(createPipeStatement.getSourceAttributes())
-
.setProcessorAttributes(createPipeStatement.getProcessorAttributes())
-
.setConnectorAttributes(createPipeStatement.getSinkAttributes());
- final TSStatus tsStatus = configNodeClient.createPipe(req);
- if (TSStatusCode.SUCCESS_STATUS.getStatusCode() != tsStatus.getCode())
{
- future.setException(new IoTDBException(tsStatus));
- } else {
- future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
- }
+ future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
}
} catch (final Exception e) {
future.setException(e);
}
return future;
}
+ private void validatePipePlugin(
+ final String pipeName,
+ final Map<String, String> sourceAttributes,
+ final Map<String, String> processorAttributes,
+ final Map<String, String> sinkAttributes)
+ throws Exception {
+ PipeDataNodeAgent.plugin()
+ .validate(
+ pipeName,
+ cloneAttributes(sourceAttributes),
+ cloneAttributes(processorAttributes),
+ cloneAttributes(sinkAttributes));
+ }
+
+ private boolean isDoubleLivingPipe(final PipeParameters
sourcePipeParameters) {
+ return sourcePipeParameters.getBooleanOrDefault(
+ Arrays.asList(
+ PipeSourceConstant.EXTRACTOR_MODE_DOUBLE_LIVING_KEY,
+ PipeSourceConstant.SOURCE_MODE_DOUBLE_LIVING_KEY),
+ PipeSourceConstant.EXTRACTOR_MODE_DOUBLE_LIVING_DEFAULT_VALUE);
+ }
+
+ private PipeParameters cloneSourceParametersWithDialect(
+ final PipeParameters sourcePipeParameters, final String sqlDialect) {
+ final Map<String, String> sourceAttributes = new
HashMap<>(sourcePipeParameters.getAttribute());
+ sourceAttributes.put(SystemConstant.SQL_DIALECT_KEY, sqlDialect);
+ return new PipeParameters(sourceAttributes);
+ }
+
+ private Map<String, String> cloneAttributes(final Map<String, String>
attributes) {
+ return new HashMap<>(attributes == null ? Collections.emptyMap() :
attributes);
+ }
+
+ private TSStatus createPipeInternal(
+ final ConfigNodeClient configNodeClient,
+ final CreatePipeStatement createPipeStatement,
+ final PipeParameters sourcePipeParameters,
+ final PipeParameters sinkPipeParameters)
+ throws TException, IllegalPathException {
+ final String pipeName = createPipeStatement.getPipeName();
+ if (PipeConfig.getInstance().getPipeAutoSplitFullEnabled()
+ && PipeDataNodeAgent.task().isFullSync(sourcePipeParameters)) {
+ // 1. Send request to create the real-time data synchronization pipeline
+ final TCreatePipeReq realtimeReq =
+ new TCreatePipeReq()
+ // Append suffix to the pipeline name for real-time data
+ .setPipeName(pipeName + "_realtime")
+ // NOTE: set if not exists always to true to handle partial
failure
+ .setIfNotExistsCondition(true)
+ // Use extractor parameters for real-time data
+ .setExtractorAttributes(
+ sourcePipeParameters
+ .addOrReplaceEquivalentAttributesWithClone(
+ new PipeParameters(
+ ImmutableMap.of(
+ SystemConstant.SQL_DIALECT_KEY,
+ sourcePipeParameters.getStringOrDefault(
+ SystemConstant.SQL_DIALECT_KEY,
+ SystemConstant.SQL_DIALECT_TREE_VALUE),
+
PipeSourceConstant.EXTRACTOR_REALTIME_ENABLE_KEY,
+ Boolean.toString(true),
+
PipeSourceConstant.EXTRACTOR_HISTORY_ENABLE_KEY,
+ Boolean.toString(false))))
+ .getAttribute())
+
.setProcessorAttributes(createPipeStatement.getProcessorAttributes())
+ .setConnectorAttributes(createPipeStatement.getSinkAttributes());
+
+ final TSStatus realtimeTsStatus =
configNodeClient.createPipe(realtimeReq);
+ // If creation fails, immediately return with exception
+ // If the procedure is still running, it's probably stuck on DataNode
+ // The pipe creation can ignore this situation and succeed, thus we do
not need to skip in
+ // this case
+ if (TSStatusCode.SUCCESS_STATUS.getStatusCode() !=
realtimeTsStatus.getCode()
+ && TSStatusCode.OVERLAP_WITH_EXISTING_TASK.getStatusCode()
+ != realtimeTsStatus.getCode()) {
+ return realtimeTsStatus;
+ }
+
+ // 2. Send request to create the historical data synchronization pipeline
+ final Map<String, String> historySinkAttributes =
+ sinkPipeParameters.hasAnyAttributes(
+ PipeSinkConstant.SINK_ENABLE_SEND_TSFILE_LIMIT,
+ PipeSinkConstant.CONNECTOR_ENABLE_SEND_TSFILE_LIMIT)
+ ? createPipeStatement.getSinkAttributes()
+ : sinkPipeParameters
+ .addOrReplaceEquivalentAttributesWithClone(
+ new PipeParameters(
+ Collections.singletonMap(
+ PipeSinkConstant.SINK_ENABLE_SEND_TSFILE_LIMIT,
+ Boolean.TRUE.toString())))
+ .getAttribute();
+
+ final TCreatePipeReq historyReq =
+ new TCreatePipeReq()
+ // Append suffix to the pipeline name for historical data
+ .setPipeName(pipeName + "_history")
+
.setIfNotExistsCondition(createPipeStatement.hasIfNotExistsCondition())
+ // Use source parameters for historical data
+ .setExtractorAttributes(
+ sourcePipeParameters
+ .addOrReplaceEquivalentAttributesWithClone(
+ new PipeParameters(
+ ImmutableMap.of(
+ SystemConstant.SQL_DIALECT_KEY,
+ sourcePipeParameters.getStringOrDefault(
+ SystemConstant.SQL_DIALECT_KEY,
+ SystemConstant.SQL_DIALECT_TREE_VALUE),
+
PipeSourceConstant.EXTRACTOR_REALTIME_ENABLE_KEY,
+ Boolean.toString(false),
+
PipeSourceConstant.EXTRACTOR_HISTORY_ENABLE_KEY,
+ Boolean.toString(true),
+ PipeSourceConstant.EXTRACTOR_MODE_KEY,
+
PipeSourceConstant.EXTRACTOR_MODE_SNAPSHOT_VALUE,
+ // We force the historical pipe to transfer
data (and maybe
+ // deletion) only
+ // Thus we can transfer schema only once
+ // And may drop the historical pipe on
successfully transferred
+ PipeSourceConstant.SOURCE_INCLUSION_KEY,
+ DataRegionListeningFilter
+
.parseInsertionDeletionListeningOptionPair(
+ sourcePipeParameters)
+ .getRight()
+ ? "data"
+ :
PipeSourceConstant.EXTRACTOR_INCLUSION_DEFAULT_VALUE,
+ PipeSourceConstant.SOURCE_EXCLUSION_KEY,
+
PipeSourceConstant.EXTRACTOR_EXCLUSION_DEFAULT_VALUE)))
+ .getAttribute())
+
.setProcessorAttributes(createPipeStatement.getProcessorAttributes())
+ .setConnectorAttributes(historySinkAttributes);
+
+ final TSStatus historyTsStatus = configNodeClient.createPipe(historyReq);
+ // If creation fails, immediately return with exception
+ if (TSStatusCode.SUCCESS_STATUS.getStatusCode() !=
historyTsStatus.getCode()) {
+ return historyTsStatus;
+ }
Review Comment:
The realtime pipe is not dropped?
--
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]
