private void parseNodeRelation(NodeRelation relation, Map<String, Node> nodeMap,
            Map<String, NodeRelation> relationMap) {
        log.info("start parse node relation, relation:{}", relation);
        Preconditions.checkNotNull(relation, "relation is null");
        Preconditions.checkState(relation.getInputs().size() > 0,
                "relation must have at least one input node");
        Preconditions.checkState(relation.getOutputs().size() > 0,
                "relation must have at least one output node");
        relation.getOutputs().forEach(s -> {
            Preconditions.checkNotNull(s, "node id in outputs is null");
            Node outputNode = nodeMap.get(s);
            Preconditions.checkNotNull(outputNode, "can not find any node by node id " + s);
            parseInputNodes(relation, nodeMap, relationMap);
            parseSingleNode(outputNode, relation, nodeMap);
            // for Load node we need to generate insert sql
            if (outputNode instanceof LoadNode) {
                insertSqls.add(genLoadNodeInsertSql((LoadNode) outputNode, relation, nodeMap));
            }
        });
        log.info("parse node relation success, relation:{}", relation);
    }
parse node relation
Here we only parse the output node in the relation,
and the input node parsing is achieved by parsing the dependent node parsing of the output node.

@param relation Define relations between nodes, it also shows the data flow
@param nodeMap Store the mapping relation between node id and node
@param relationMap Store the mapping relation between node id and relation
inlong-sort/sort-core/src/main/java/org/apache/inlong/sort/parser/impl/FlinkSqlParser.java