private String genUnionNodeSelectSql(Node node, List<FieldRelation> fieldRelations,
            UnionNodeRelation unionRelation, Map<String, Node> nodeMap) {
        Preconditions.checkState(unionRelation.getInputs().size() > 1,
                "union must have more than one input nodes");
        Preconditions.checkState(unionRelation.getOutputs().size() == 1,
                "union node only support one output node");
        // Get table name alias map by input nodes
        Map<String, Map<String, FieldRelation>> fieldRelationMap = new HashMap<>(unionRelation.getInputs().size());
        // Generate mapping for output field to FieldRelation
        fieldRelations.forEach(s -> {
            // All field relations of input nodes will be the same if the node id of output field is blank.
            // Currently, the node id in the output field is used to distinguish which field of the node in the
            // upstream of the union the field comes from. A better way is through the upstream input field,
            // but this abstraction does not yet have the ability to set node ids for all upstream input fields.
            // todo optimize the implementation of this block in the future
            String nodeId = s.getOutputField().getNodeId();
            if (StringUtils.isBlank(nodeId)) {
                nodeId = unionRelation.getInputs().get(0);
            }
            fieldRelationMap.computeIfAbsent(nodeId, k -> new HashMap<>()).put(s.getOutputField().getName(), s);
        });
        StringBuilder sb = new StringBuilder();
        sb.append(genUnionSingleSelectSql(unionRelation.getInputs().get(0),
                nodeMap.get(unionRelation.getInputs().get(0)).genTableName(), node.getFields(),
                fieldRelationMap, fieldRelationMap.get(unionRelation.getInputs().get(0)), node));
        String relationFormat = unionRelation.format();
        for (int i = 1; i < unionRelation.getInputs().size(); i++) {
            String inputId = unionRelation.getInputs().get(i);
            sb.append("\n").append(relationFormat).append("\n")
                    .append(genUnionSingleSelectSql(inputId, nodeMap.get(inputId).genTableName(), node.getFields(),
                            fieldRelationMap, fieldRelationMap.get(unionRelation.getInputs().get(0)), node));
        }
        return sb.toString();
    }
Generate select sql for union

@param fieldRelations The relation of fieds
@param unionRelation The union relation of sql
@param nodeMap Store the mapping relation between node id and node
@return Transform sql for this node logic
inlong-sort/sort-core/src/main/java/org/apache/inlong/sort/parser/impl/FlinkSqlParser.java