private String genTransformSelectSql(TransformNode transformNode, NodeRelation relation,
            Map<String, Node> nodeMap) {
        String selectSql;
        if (relation instanceof JoinRelation) {
            // parse join relation and generate the transform sql
            JoinRelation joinRelation = (JoinRelation) relation;
            selectSql = genJoinSelectSql(transformNode, transformNode.getFieldRelations(), joinRelation,
                    transformNode.getFilters(), transformNode.getFilterStrategy(), nodeMap);
        } else if (relation instanceof UnionNodeRelation) {
            // parse union relation and generate the transform sql
            Preconditions.checkState(transformNode.getFilters() == null
                    || transformNode.getFilters().isEmpty(), "Filter is not supported when union");
            Preconditions.checkState(transformNode.getClass() == TransformNode.class,
                    String.format("union is not supported for %s", transformNode.getClass().getSimpleName()));
            UnionNodeRelation unionRelation = (UnionNodeRelation) relation;
            selectSql = genUnionNodeSelectSql(transformNode, transformNode.getFieldRelations(), unionRelation, nodeMap);
        } else {
            // parse base relation that one to one and generate the transform sql
            Preconditions.checkState(relation.getInputs().size() == 1,
                    "simple transform only support one input node");
            Preconditions.checkState(relation.getOutputs().size() == 1,
                    "join node only support one output node");
            selectSql = genSimpleSelectSql(transformNode, transformNode.getFieldRelations(), relation,
                    transformNode.getFilters(), transformNode.getFilterStrategy(), nodeMap);
        }
        return selectSql;
    }
Parse a single node and generate the corresponding sql

@param node The abstract of extract, transform, load
@param relation Define relations between nodes, it also shows the data flow
@param nodeMap store the mapping relation between node id and node
inlong-sort/sort-core/src/main/java/org/apache/inlong/sort/parser/impl/FlinkSqlParser.java