private String genLoadSelectSql(LoadNode loadNode, NodeRelation relation,
            Map<String, Node> nodeMap) {
        String selectSql;
        if (relation instanceof JoinRelation) {
            // parse join relation and generate the select sql
            JoinRelation joinRelation = (JoinRelation) relation;
            selectSql = genJoinSelectSql(loadNode, loadNode.getFieldRelations(), joinRelation,
                    loadNode.getFilters(), loadNode.getFilterStrategy(), nodeMap);
        } else if (relation instanceof UnionNodeRelation) {
            // parse union relation and generate the select sql
            Preconditions.checkState(loadNode.getFilters() == null || loadNode.getFilters().isEmpty(),
                    "Filter is not supported when union");
            UnionNodeRelation unionRelation = (UnionNodeRelation) relation;
            selectSql = genUnionNodeSelectSql(loadNode, loadNode.getFieldRelations(), unionRelation, nodeMap);
        } else {
            // parse base relation that one to one and generate the select 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(loadNode, loadNode.getFieldRelations(), relation,
                    loadNode.getFilters(), loadNode.getFilterStrategy(), nodeMap);
        }
        return selectSql;
    }
Generate load node insert sql

@param loadNode The real data write node
@param relation The relation between nods
@param nodeMap The node map
@return Insert sql
inlong-sort/sort-core/src/main/java/org/apache/inlong/sort/parser/impl/FlinkSqlParser.java