private String genJoinSelectSql(Node node, List<FieldRelation> fieldRelations,
            JoinRelation relation, List<FilterFunction> filters, FilterStrategy filterStrategy,
            Map<String, Node> nodeMap) {
        Preconditions.checkState(relation.getInputs().size() > 1,
                "join must have more than one input nodes");
        Preconditions.checkState(relation.getOutputs().size() == 1,
                "join node only support one output node");
        // Get table name alias map by input nodes
        Map<String, String> tableNameAliasMap = new HashMap<>(relation.getInputs().size());
        relation.getInputs().forEach(s -> {
            Node inputNode = nodeMap.get(s);
            Preconditions.checkNotNull(inputNode, String.format("input node is not found by id:%s", s));
            tableNameAliasMap.put(s, String.format("t%s", s));
        });
        StringBuilder sb = new StringBuilder();
        sb.append("SELECT ");
        Map<String, FieldRelation> fieldRelationMap = new HashMap<>(fieldRelations.size());
        // Generate mapping for output field to FieldRelation
        fieldRelations.forEach(s -> {
            fillOutTableNameAlias(Collections.singletonList(s.getInputField()), tableNameAliasMap);
            fieldRelationMap.put(s.getOutputField().getName(), s);
        });

        if (node instanceof HbaseLoadNode) {
            HbaseLoadNode hbaseLoadNode = (HbaseLoadNode) node;
            parseHbaseLoadFieldRelation(hbaseLoadNode.getRowKey(), hbaseLoadNode.getFieldRelations(), sb);
        } else {
            parseFieldRelations(node.getFields(), fieldRelationMap, sb);
        }
        if (node instanceof DistinctNode) {
            DistinctNode distinctNode = (DistinctNode) node;
            // Fill out the tablename alias for param
            List<FunctionParam> params = new ArrayList<>(distinctNode.getDistinctFields());
            params.add(distinctNode.getOrderField());
            fillOutTableNameAlias(params, tableNameAliasMap);
            // Generate distinct sql, such as ROW_NUMBER()...
            genDistinctSql(distinctNode, sb);
        }
        sb.append(" FROM `").append(nodeMap.get(relation.getInputs().get(0)).genTableName()).append("` ")
                .append(tableNameAliasMap.get(relation.getInputs().get(0)));
        // Parse condition map of join and format condition to sql, such as on 1 = 1...
        Map<String, List<FilterFunction>> conditionMap = relation.getJoinConditionMap();
        if (relation instanceof TemporalJoinRelation) {
            parseTemporalJoin((TemporalJoinRelation) relation, nodeMap, tableNameAliasMap, conditionMap, sb);
        } else if (relation instanceof IntervalJoinRelation) {
            Preconditions.checkState(filters == null || filters.isEmpty(),
                    String.format("filters must be empty for %s", relation.getClass().getSimpleName()));
            parseIntervalJoin((IntervalJoinRelation) relation, nodeMap, tableNameAliasMap, sb);
            List<FilterFunction> conditions = conditionMap.values().stream().findFirst().orElse(null);
            Preconditions.checkState(conditions != null && !conditions.isEmpty(),
                    String.format("Join conditions must no be empty for %s", relation.getClass().getSimpleName()));
            fillOutTableNameAlias(new ArrayList<>(conditions), tableNameAliasMap);
            parseFilterFields(FilterStrategy.RETAIN, conditions, sb);
        } else {
            parseRegularJoin(relation, nodeMap, tableNameAliasMap, conditionMap, sb);
        }
        if (filters != null && !filters.isEmpty()) {
            // Fill out the table name alias for param
            fillOutTableNameAlias(new ArrayList<>(filters), tableNameAliasMap);
            // Parse filter fields to generate filter sql like 'WHERE 1=1...'
            parseFilterFields(filterStrategy, filters, sb);
        }
        if (node instanceof DistinctNode) {
            // Generate distinct filter sql like 'WHERE row_num = 1'
            sb = genDistinctFilterSql(node.getFields(), sb);
        }
        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