public static void optimizeNodeRelation(StreamInfo streamInfo, List<TransformResponse> transformResponses) {
        if (CollectionUtils.isEmpty(transformResponses)) {
            return;
        }
        Map<String, TransformDefinition> transformTypeMap = transformResponses.stream().collect(
                Collectors.toMap(TransformResponse::getTransformName, transformResponse -> {
                    TransformType transformType = TransformType.forType(transformResponse.getTransformType());
                    return StreamParseUtils.parseTransformDefinition(transformResponse.getTransformDefinition(),
                            transformType);
                }));
        List<Node> nodes = streamInfo.getNodes();
        Map<String, TransformNode> joinNodes = nodes.stream().filter(node -> node instanceof TransformNode)
                .map(node -> (TransformNode) node)
                .filter(transformNode -> {
                    TransformDefinition transformDefinition = transformTypeMap.get(transformNode.getName());
                    return JOIN_NODES.contains(transformDefinition.getTransformType());
                }).collect(Collectors.toMap(TransformNode::getName, transformNode -> transformNode));

        List<NodeRelation> relations = streamInfo.getRelations();
        Iterator<NodeRelation> shipIterator = relations.listIterator();
        List<NodeRelation> joinRelations = Lists.newArrayList();
        while (shipIterator.hasNext()) {
            NodeRelation relation = shipIterator.next();
            List<String> outputs = relation.getOutputs();
            if (outputs.size() == 1) {
                String nodeName = outputs.get(0);
                if (joinNodes.get(nodeName) != null) {
                    TransformDefinition transformDefinition = transformTypeMap.get(nodeName);
                    TransformType transformType = transformDefinition.getTransformType();
                    switch (transformType) {
                        case JOINER:
                            joinRelations.add(getNodeRelation((JoinerDefinition) transformDefinition, relation));
                            shipIterator.remove();
                            break;
                        case LOOKUP_JOINER:
                            assert transformDefinition instanceof LookUpJoinerDefinition;
                            joinRelations.add(getNodeRelation((LookUpJoinerDefinition) transformDefinition, relation));
                            shipIterator.remove();
                            break;
                        case INTERVAL_JOINER:
                            if (transformDefinition instanceof IntervalJoinerDefinition) {
                                joinRelations
                                        .add(getNodeRelation((IntervalJoinerDefinition) transformDefinition, relation));
                            }
                            shipIterator.remove();
                            break;
                        case TEMPORAL_JOINER:
                            assert transformDefinition instanceof TemporalJoinerDefinition;
                            joinRelations
                                    .add(getNodeRelation((TemporalJoinerDefinition) transformDefinition, relation));
                            shipIterator.remove();
                            break;
                        default:
                            throw new IllegalArgumentException(
                                    String.format("Unsupported transformType for %s", transformType));
                    }
                }
            }
        }
        relations.addAll(joinRelations);
    }
Optimize relation of node, JoinerRelation must be rebuilt.
inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/sort/util/NodeRelationUtils.java