private void parseFieldRelations(List<FieldInfo> fields,
            Map<String, FieldRelation> fieldRelationMap, StringBuilder sb) {
        for (FieldInfo field : fields) {
            FieldRelation fieldRelation = fieldRelationMap.get(field.getName());
            FormatInfo fieldFormatInfo = field.getFormatInfo();
            if (fieldRelation == null) {
                String targetType = TableFormatUtils.deriveLogicalType(fieldFormatInfo).asSummaryString();
                sb.append("\n    CAST(NULL as ").append(targetType).append(") AS ").append(field.format()).append(",");
                continue;
            }
            boolean complexType = fieldFormatInfo instanceof RowFormatInfo
                    || fieldFormatInfo instanceof ArrayFormatInfo
                    || fieldFormatInfo instanceof MapFormatInfo;
            FunctionParam inputField = fieldRelation.getInputField();
            if (inputField instanceof FieldInfo) {
                FieldInfo fieldInfo = (FieldInfo) inputField;
                FormatInfo formatInfo = fieldInfo.getFormatInfo();
                FieldInfo outputField = fieldRelation.getOutputField();
                boolean sameType = formatInfo != null
                        && outputField != null
                        && outputField.getFormatInfo() != null
                        && outputField.getFormatInfo().getTypeInfo().equals(formatInfo.getTypeInfo());
                if (complexType || sameType || fieldFormatInfo == null) {
                    sb.append("\n    ").append(inputField.format()).append(" AS ").append(field.format()).append(",");
                } else {
                    String targetType = TableFormatUtils.deriveLogicalType(fieldFormatInfo).asSummaryString();
                    sb.append("\n    CAST(").append(inputField.format()).append(" as ")
                            .append(targetType).append(") AS ").append(field.format()).append(",");
                }
            } else {
                String targetType = TableFormatUtils.deriveLogicalType(field.getFormatInfo()).asSummaryString();
                sb.append("\n    CAST(").append(inputField.format()).append(" as ")
                        .append(targetType).append(") AS ").append(field.format()).append(",");
            }
        }
        sb.deleteCharAt(sb.length() - 1);
    }
Parse field relation

@param fields The fields defined in node
@param fieldRelationMap The field relation map
@param sb Container for storing sql
inlong-sort/sort-core/src/main/java/org/apache/inlong/sort/parser/impl/FlinkSqlParser.java