private void parseInputNodes(NodeRelation relation, Map<String, Node> nodeMap,
            Map<String, NodeRelation> relationMap) {
        for (String upstreamNodeId : relation.getInputs()) {
            if (!hasParsedSet.contains(upstreamNodeId)) {
                Node upstreamNode = nodeMap.get(upstreamNodeId);
                Preconditions.checkNotNull(upstreamNode,
                        "can not find any node by node id " + upstreamNodeId);
                parseSingleNode(upstreamNode, relationMap.get(upstreamNodeId), nodeMap);
            }
        }
    }
parse the input nodes corresponding to 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