nzw921rx commented on PR #10941:
URL: https://github.com/apache/seatunnel/pull/10941#issuecomment-4529150131
> > @dybyte good job , i have a quesion
> > could the copy be handled at the `TransformFlowLifeCycle` level rather
than inside each individual transform? This way, individual transforms wouldn't
need to worry about upstream data sharing behavior.
>
> Considering there is no `transform` scenario.
@davidzollo Thank you very much for your attention to my question. Without
transforms, `TransformFlowLifeCycle` will not execute any copy logic — there is
already an` if (transform.isEmpty())` guard in the code that returns early. So
the copy would only be needed when the transform list is non-empty, which is
exactly when `TransformFlowLifeCycle` should handle it uniformly.
The following is the source code:
```java
@Override
public void received(Record<?> record) {
if (record.getData() instanceof Barrier) {
} else if (record.getData() instanceof SchemaChangeEvent) {
} else {
if (prepareClose) {
return;
}
T inputData = (T) record.getData();
List<T> outputDataList = transform(inputData);
if (!outputDataList.isEmpty()) {
// todo log metrics
for (T outputData : outputDataList) {
collector.collect(new Record<>(outputData));
}
}
}
}
public List<T> transform(T inputData) {
// Considering there is no transform scenario.
if (transform.isEmpty()) {
return Collections.singletonList(inputData);
}
List<T> dataList = new ArrayList<>();
dataList.add(inputData);
for (SeaTunnelTransform<T> transformer : transform) {
List<T> nextInputDataList = new ArrayList<>();
if (transformer instanceof SeaTunnelFlatMapTransform) {
SeaTunnelFlatMapTransform<T> transformDecorator =
(SeaTunnelFlatMapTransform<T>) transformer;
for (T data : dataList) {
List<T> outputDataArray =
transformDecorator.flatMap(data);
log.debug(
"Transform[{}] input row {} and output row {}",
transformer,
data,
outputDataArray);
if (CollectionUtils.isNotEmpty(outputDataArray)) {
nextInputDataList.addAll(outputDataArray);
}
}
} else if (transformer instanceof SeaTunnelMapTransform) {
for (T data : dataList) {
SeaTunnelMapTransform<T> transformDecorator =
(SeaTunnelMapTransform<T>) transformer;
T outputData = transformDecorator.map(data);
log.debug(
"Transform[{}] input row {} and output row {}",
transformer,
data,
outputData);
if (outputData == null) {
log.trace("Transform[{}] filtered data row {}",
transformer, data);
continue;
}
nextInputDataList.add(outputData);
}
}
dataList = nextInputDataList;
}
return dataList;
}
```
--
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]
